SQLAlchemy命令式映射复合实体：无需__composite_values__及第三方依赖

方案1：自定义TypeDecorator封装复合类型转换 #

用SQLAlchemy的TypeDecorator写一个自定义类型，专门处理Money对象和数据库两列的双向转换，完全不用改动原有的Money和Transaction类。

代码示例 #

from sqlalchemy import TypeDecorator, Integer, String, Column, BigInteger, DateTime, text
from sqlalchemy.sql import expression
from sqlalchemy.orm import registry
from dataclasses import dataclass
import datetime

mapper_registry = registry()

@dataclass(kw_only=True)
class Money:
    amount: int
    currency: str

@dataclass(kw_only=True)
class Transaction:
    id: int
    value: Money
    description: str
    timestamp: datetime.datetime

class MoneyType(TypeDecorator):
    impl = expression.ColumnClauseList(Integer, String)
    cache_ok = True

    def process_bind_param(self, value, dialect):
        if not value:
            return (None, None)
        return (value.amount, value.currency)

    def process_result_value(self, value, dialect):
        if not value:
            return None
        amount, currency = value
        return Money(amount=amount, currency=currency)

transaction_table = Table(
    "transaction",
    mapper_registry.metadata,
    Column("id", BigInteger, primary_key=True),
    Column("description", String(1024)),
    Column(
        "timestamp",
        DateTime(timezone=False),
        nullable=False,
        server_default=text("NOW()"),
    ),
    # 用自定义类型关联两个实际列
    Column("value", MoneyType, (
        Column("value_amount", Integer(), nullable=False),
        Column("value_currency", String(5), nullable=False),
    ))
)

mapper_registry.map_imperatively(
    Transaction,
    transaction_table,
    properties={"value": transaction_table.c.value}
)

方案2：手动实现属性的getter/setter #

在命令式映射里，直接给value属性定义读写逻辑，绕过composite函数对__composite_values__的强制要求。读写value时自动和value_amount、value_currency列做转换。

代码示例 #

mapper_registry.map_imperatively(
    Transaction,
    transaction_table,
    properties={
        "value": property(
            # 读取时从列数据构造Money对象
            lambda obj: Money(amount=obj.value_amount, currency=obj.value_currency),
            # 赋值时把Money的属性拆分到对应列
            lambda obj, val: (setattr(obj, "value_amount", val.amount), setattr(obj, "value_currency", val.currency))
        )
    },
    # 要把底层的两个列也加入映射，不然实例访问不到
    include_properties=["id", "description", "timestamp", "value_amount", "value_currency"]
)

方案3：用CompositeType+自定义expander参数 #

用SQLAlchemy的CompositeType定义数据库层面的复合类型，然后在composite映射时，通过expander参数指定如何从Money对象提取字段，不用给原类加__composite_values__方法。

代码示例 #

from sqlalchemy import CompositeType

# 定义数据库复合类型
money_type = CompositeType(
    "money_type",
    [
        Column("amount", Integer),
        Column("currency", String(5))
    ]
)

transaction_table = Table(
    "transaction",
    mapper_registry.metadata,
    Column("id", BigInteger, primary_key=True),
    Column("description", String(1024)),
    Column(
        "timestamp",
        DateTime(timezone=False),
        nullable=False,
        server_default=text("NOW()"),
    ),
    Column("value", money_type, nullable=False)
)

mapper_registry.map_imperatively(
    Transaction,
    transaction_table,
    properties={
        "value": composite(
            Money,
            transaction_table.c.value.amount,
            transaction_table.c.value.currency,
            # 自定义提取字段的逻辑，替代__composite_values__
            expander=lambda money_obj: (money_obj.amount, money_obj.currency)
        )
    }
)

内容的提问来源于stack exchange，提问作者Andrew Lalis