在SQLAlchemy的Mapped Class构造函数中有忽略额外关键字的选项吗？

Question

根据以下内容，我正在尝试从一个具有额外键的Python字典中初始化SQLAlchemy映射类。是否可能让映射类自动忽略额外的键而不是抛出错误？同样地，如果缺少键，映射类是否可以有默认值？

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)

哪个会引发无效密钥错误

有什么想法？

- mgcdanny

Answer 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)

- Eric Ihli

Answer 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），但如果犯了错误，可能不容易发现问题。

- uvsmtid

---

哇，这真的很有帮助。谢谢。 - Andrew Clark

Answer 3

如果您的模型有关联，您可以使用模型的Mapper对象，就像@eric-ihli提到的那样。这是另一种方法（请注意__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()

- pmsoltani

Answer 4

ORM模型的默认构造函数如下所示（为了简洁起见，已删除文档字符串）：

def _declarative_constructor(self: Any, **kwargs: Any) -> None:
    cls_ = type(self)
    for k in kwargs:
        if not hasattr(cls_, k):
            raise TypeError(
                "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
            )
        setattr(self, k, kwargs[k])

所以在其他答案中，很容易定义一个__init__方法来扩展Base。

def __init__(self, **kwargs):
    super().__init__(**{k: v for k, v in kwargs.items() if hasattr(type(self), k)})

然而，我们可以通过直接替换默认构造函数来省去显式的__init__的需要。给定这个函数：

def lenient_constructor(self, **kwargs):
    cls_ = type(self)
    for k in kwargs:
        if not hasattr(cls_, k):
            print(f'Skipping invalid attr {k!r}')
            continue
        setattr(self, k, kwargs[k])

然后对于2.0之前的语法，我们可以将其设置为Base的默认构造函数，就像这样：

from sqlalchemy import orm

Base = orm.declarative_base(constructor=lenient_constructor)

或者像这样为2.0+版本：

from sqlalchemy import orm

registry = orm.registry(constructor=lenient_constructor)


class Base(orm.DeclarativeBase):
    registry = registry

- snakecharmerb

Answer 5

我们是否保证超类的__init__仅设置__dict__条目，而不会产生其他所需效果？我完全跳过超类调用感到有些不舒服，因此我尝试解决此问题的方式是，仅传递与列名对应的条目：

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)

- nochmal

Answer 6

此外，为了传递额外的关键词并调用Base.__init__()方法，您可以从super()中排除extrakeys，然后做您想做的事情：

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']

- R Yakovlev

Answer 7

根据 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 一样，但不必每次使用时都调用它。

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

- Bruce Edge

Answer 8

这是我解决类似问题的方法，基本上是检查ORM模型的列，并根据它来过滤字典。

from sqlalchemy.inspection import inspect
from sqlalchemy import Column, Integer, String

class User(Base):
     __tablename__ = 'users'

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

VALID_COLUMNS = [column.key for column in inspect(User).columns]
my_example_user = {'id'=1, 'name'='john', 'extra_key'= 1234}
my_example_user = {k: v for k, v in my_example_user.items() if k in VALID_COLUMNS}

User(**my_example_user)

更多信息请参考https://docs.sqlalchemy.org/en/20/core/inspection.html。

希望对您有所帮助。

- Sanju