在 sqlalchemy 映射类构造函数中忽略额外关键字的选项？

Posted 2023-04-12

【中文标题】在 sqlalchemy 映射类构造函数中忽略额外关键字的选项？

【英文标题】：Option to ignore extra keywords in an sqlalchemy Mapped Class constructor?

【发布时间】：2015-11-18 21:29:48

【问题描述】：

如下所示，我正在尝试从具有额外键的 python 字典中初始化一个 sqlalchemy 映射类。是否可以让 Mapped Class 自动忽略额外的键而不是抛出错误？同样，如果键不存在，映射类可以有默认值吗？

from sqlalchemy import Column, Integer, String
class User(Base):
     __tablename__ = 'users'

     id = Column(Integer, primary_key=True)
     name = Column(String)

这是初始化部分：

my_example_user = 'id'=1, 'name'='john', 'extra_key'= 1234
User(**my_example_user)

这会引发无效的密钥错误

想法？

【参考方案1】：

SQLAlchemy Mapper 对象有一个 attrs 属性，它是映射类字段名称的字典。

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import class_mapper
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class User(Base):
    __tablename__ = 'user'

    id = Column(Integer, primary_key=True)
    name = Column(String)

user = 
    'name': 'Eihli',
    'skill': 11


user_mapper = class_mapper(User)
mapped_user = User(**user)
# Boom! TypeError: 'skill' is an invalid keyword argument for User

mapped_user = User(**
    k: v for k, v in user.items()
    if k in user_mapper.attrs.keys()
)
# Success!

无需费心维护排除列表或使用 dict 或妨碍超级调用。

如果您尝试使用嵌套数据生成模型，则必须做一些不同的事情。否则你会得到一个“Unhashable type 'dict'”错误。

这是一个检查映射器并获取关系键的帮助器示例。

def from_json(model, data):
    mapper = class_mapper(model)
    keys = mapper.attrs.keys()
    relationships = inspect(mapper).relationships
    args = k: v for k, v in data.items()
            if k in keys and k not in relationships
    return model(**args)

【参考方案2】：

简而言之，定义不将参数传递给其超类的构造函数：

class User(Base):

    # ...

    def __init__(self, **entries):

        # NOTE: Do not call superclass
        #       (which is otherwise a default behaviour).
        #super(User, self).__init__(**entries)

        self.__dict__.update(entries)

我在从peewee 转换时遇到了同样的问题，这需要相反的 - 将参数传递给它的超类（因此，构造函数已经定义）。所以，我只是试着注释掉这条线，事情就开始起作用了。

更新

此外，请确保entries 不包含（并因此覆盖）为 SQLAlchemy 定义的 User 类中的任何元字段，例如那些 ORM 关系。这有点明显（SQLAlchemy），但是一旦出错，可能不容易发现问题。

【讨论】：

哇，这太有帮助了。谢谢

【参考方案3】：

除了设置__dict__ 条目之外，我们是否保证现有超类的__init__ 永远不会有其他预期效果？完全绕过超类调用我感觉不太舒服，所以我尝试解决这个问题如下，只传递与列名对应的条目：

class User(Base):

    # ...

    def __init__(self, **entries):
        '''Override to avoid TypeError when passed spurious column names'''
        col_names = set([col.name for col in self.__table__.columns])
        superentries = k : entries[k] for k in col_names.intersection(entries.keys())
        super().__init__(**superentries)

【参考方案4】：

还可以传递额外的关键字并调用Base.__init__() 方法，您可以从super() 中排除额外的关键字，然后执行您想要的操作：

from sqlalchemy import Column, Integer, String

class User(Base):
     __tablename__ = 'users'

     id = Column(Integer, primary_key=True)
     name = Column(String)

     def __init__(self, **kwargs):
         extra_kw_list = ['key1', 'key2']
         super(User, self).__init__(**x: y for x, y in kwargs.items()
                                       if x not in extra_kw_list)
         #do something you need here
         item1, item2 = kwargs['key1'], kwargs['key2']

【参考方案5】：

根据 R Yakovlev 的回答，您可以使元素列表动态化：

from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    def __init__(self, **kwargs):
        keep_kwargs = k: v for k, v in kwargs.items() if k in user_columns
        super(User, self).__init__(**keep_kwargs)


user_columns = [_ for _ in User.__dict__.keys() if not _.startswith('_')]

我想尝试找到一种方法将 user_columns 嵌入到对象中，就像使用 @hybrid_property 一样，但不要在每次使用时都调用它。

我认为这是可能的，但超出了我的时间限制。

【参考方案6】：

如果您的模型有关系，您可以使用模型的 Mapper 对象，如 @eric-ihli mentioned。这是另一种方式（注意__init__ 方法）：

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import backref, relationship

from my_app.db_models import Base


class Employee(Base):
    __tablename__ = "employee"

    id = Column(Integer, primary_key=True, autoincrement=True)
    department_id = Column(Integer, ForeignKey("department.id"), index=True)

    email = Column(String, unique=True, index=True, nullable=False)
    name = Column(String)

    department = relationship(
        "Department", backref=backref("employees", cascade="all, delete-orphan")
    )


    def __init__(self, **kwargs):
        allowed_args = self.__mapper__.class_manager  # returns a dict
        kwargs = k: v for k, v in kwargs.items() if k in allowed_args
        super().__init__(**kwargs)

这样，您可以像这样创建员工模型：

from contextlib import closing
from my_app.db_models import Department, Employee, SessionLocal


with closing(SessionLocal()) as db:
    dept = db.query(Department).filter(Department.name == 'HR').first()
    employee = Employee(name='John Smith', email='[e-mail]', department=dept)
    db.add(employee)
    db.commit()