SQLAlchemy 的问题 - 推断外键的棉花糖嵌套对象

Question

我试图让 Marshmallow-SQLAlchemy 反序列化具有嵌套对象的对象，而不指定嵌套对象的外键（应该是父对象的主键）。这是一个独立的示例：

# Python version == 3.8.2
from datetime import datetime
import re

# SQLAlchemy == 1.3.23
from sqlalchemy import func, create_engine, Column, ForeignKey, Text, DateTime
from sqlalchemy.ext.declarative import as_declarative, declared_attr
from sqlalchemy.orm import relationship, sessionmaker

# marshmallow==3.10.0
# marshmallow-sqlalchemy==0.24.2
from marshmallow import fields
from marshmallow.fields import Nested
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema

################################################################################
# Set up
################################################################################

engine = create_engine("sqlite:///test.db")

Session = sessionmaker()
Session.configure(bind=engine)
session = Session()


################################################################################
# Models
################################################################################

@as_declarative()
class Base(object):

    @declared_attr
    def __tablename__(cls):
        # From https://stackoverflow.com/questions/1175208/elegant-python-function-to-convert-camelcase-to-snake-case
        name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', cls.__name__)
        return re.sub('([a-z0-9])([A-Z])', r'\1_\2', name).lower()

    @declared_attr
    def updated(cls):
        return Column(DateTime, default=func.now(), onupdate=func.now(), nullable=False)


class Account(Base):
    id = Column(Text, primary_key=True)
    name = Column(Text, nullable=False)
    tags = relationship("AccountTag", backref="account")


class AccountTag(Base):
    account_id = Column(Text, ForeignKey('account.id'), primary_key=True)
    Key = Column(Text, primary_key=True)
    Value = Column(Text, nullable=False)


################################################################################
# Schemas
################################################################################

class AutoSchemaWithUpdate(SQLAlchemyAutoSchema):
    class Meta:
        load_instance = True
        sqla_session = session
    updated = fields.DateTime(default=lambda: datetime.now())


class AccountSchema(AutoSchemaWithUpdate):
    class Meta:
        model = Account
        include_relationships = True

    tags = Nested("AccountTagSchema", many=True)


class AccountTagSchema(AutoSchemaWithUpdate):
    class Meta:
        model = AccountTag
        include_fk = True


################################################################################
# Test
################################################################################

Base.metadata.create_all(engine)

account_object = AccountSchema().load({
        "id": "ABC1234567",
        "name": "Account Name",
        "tags": [
            {
                "Value": "Color",
                "Key": "Blue"
            }
        ]
    })

session.merge(account_object)

session.commit()

这是我得到的错误：

Traceback (most recent call last):
  File "example.py", line 88, in <module>
    account_object = AccountSchema().load({
  File "C:\python\site-packages\marshmallow_sqlalchemy\schema\load_instance_mixin.py", line 92, in load
    return super().load(data, **kwargs)
  File "C:\python\site-packages\marshmallow\schema.py", line 727, in load
    return self._do_load(
  File "C:\python\site-packages\marshmallow\schema.py", line 909, in _do_load
    raise exc
marshmallow.exceptions.ValidationError: {'tags': {0: {'account_id': ['Missing data for required field.']}}}

我觉得我正在尝试做一些直观的事情，但我不确定了。我确定我在这里很近，但没有运气让它发挥作用。非常感谢您的帮助。

Answer 1

您收到错误是因为您在 Meta 类中为 AccountTagSchema 指定了 include_fk。

您可以检查已为架构生成的字段：

print(AccountTagSchema._declared_fields["account_id"])
# <fields.String(default=<marshmallow.missing>, attribute=None, validate=[], required=True, load_only=False, dump_only=False, missing=<marshmallow.missing>, allow_none=False, error_messages={'required': 'Missing data for required field.', 'null': 'Field may not be null.', 'validator_failed': 'Invalid value.', 'invalid': 'Not a valid string.', 'invalid_utf8': 'Not a valid utf-8 string.'})>

请注意，它生成account_id 和required=True，这是由于它表示的sqlalchemy 列是NOT NULL，因为它是主键的一部分。

所以最简单的方法是从模式元中删除include_fk：

class AccountTagSchema(AutoSchemaWithUpdate):
    class Meta(AutoSchemaWithUpdate.Meta):
        model = AccountTag
        #  include_fk = True  <--- remove

...但是，运行脚本会遇到另一个问题：

sqlalchemy.orm.exc.UnmappedInstanceError: 类 'builtins.dict' 未映射

这意味着我们最终将 dict 传递给 SQLAlchemy 会话，它期待映射的 Base 子类。

这样做的原因是当子类从基本模式继承时，例如 AutoSchemaWithUpdate 在这种情况下，子类不会自动继承父类的元配置。文档提供了a couple of strategies for this，最简单的是孩子的Meta 类也应该继承自父母的Meta 类：

class AccountSchema(AutoSchemaWithUpdate):
    class Meta(AutoSchemaWithUpdate.Meta):  # <--- this here
        model = Account
        include_relationships = True

    tags = Nested("AccountTagSchema", many=True)

一旦我们为AccountSchema 和AccountTagSchema 都这样做了，我们就准备好再次运行脚本并且它可以工作了……第一次。立即再次运行脚本，又出现一个错误：

AssertionError: 依赖规则试图清除实例“”上的主键列“account_tag.account_id”

这是设计决定使加载的AccountTag 实例无法识别（即，从有效负载中排除主键）以及决定将外键字段作为AccountTag 主键的一部分的结果.

SQLAlchemy 无法识别新创建的AccountTag 实例与已经存在的实例相同，因此它首先尝试通过将外键字段的值设置为None 来解除原始帐户标签与帐户的关联.但是，这是不允许的，因为外键也是主键，不能设置为 NULL。

解决方案描述为here，涉及在relationship 上设置显式cascade：

class Account(Base):
    id = Column(Text, primary_key=True)
    name = Column(Text, nullable=False)
    tags = relationship("AccountTag", backref="account", cascade="all,delete-orphan")

现在再次运行脚本，每次都会运行。