SQLAlchemy Core映射dataclass：User与Language关联赋值问题

问题根源在于你当前将User.ui_language_code直接映射到数据库的字符串外键字段，导致只能传入字符串。要实现直接操作对象，需将该属性定义为SQLAlchemy的relationship，同时将数据库外键字段映射为内部私有属性，让SQLAlchemy自动处理外键与对象的转换。

修改步骤与完整代码 #

1. 调整映射逻辑与模型配置 #

from dataclasses import dataclass
from typing import Optional
import logging as logger

from sqlalchemy import (
    Table, MetaData, Column, BIGINT, String, ForeignKey, func, select
)
from sqlalchemy.orm import relationship, registry, sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine


DATABASE_URL = "sqlite+aiosqlite:///:memory:"
engine = create_async_engine(
    DATABASE_URL,
    # echo=True
)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


mapper_registry = registry()
metadata = MetaData()


# domain layer ---------------------------------------------------------------
@dataclass
class Language(object):
    code: str
    title: str

    def __repr__(self):
        return f"<Language {self.code} {self.title}>"


@dataclass
class User(object):
    id: int
    username: Optional[str]
    first_name: str
    last_name: Optional[str]
    ui_language_code: Language  # 业务层直接操作Language对象

    @property
    def full_name(self):
        return f"{self.first_name} {self.last_name or ''}"

# ---------------------------------------------------------------------------


# infrastructure layer ------------------------------------------------------
language_table = Table(
    "languages",
    metadata,
    Column("code", String(10), primary_key=True, unique=True),
    Column("title", String),
)


user_table = Table(
    "users",
    metadata,
    Column("id", BIGINT, primary_key=True, autoincrement=False),
    Column("username", String(32), unique=True),
    Column("first_name", String(64)),
    Column("last_name", String(64), default="", server_default=""),
    Column("ui_language_code", String, ForeignKey("languages.code", ondelete="CASCADE")),
)


def configure_mappers() -> None:
    logger.info("Configuring mappers")

    # 先映射Language，确保User关联时模型已注册
    mapper_registry.map_imperatively(
        Language,
        language_table,
    )

    mapper_registry.map_imperatively(
        User,
        user_table,
        properties={
            # 将数据库外键字段映射为私有内部属性，由SQLAlchemy自动维护
            '_ui_language_code': user_table.c.ui_language_code,
            # 业务层属性关联Language模型，自动处理外键与对象的转换
            'ui_language_code': relationship(
                Language,
                foreign_keys=[user_table.c.ui_language_code],
                uselist=False,  # 一对一关联，不返回列表
                lazy='selectin'  # 按需加载关联对象，可改为'joined'实现立即加载
            ),
        }
    )


configure_mappers()

# ---------------------------------------------------------------------------


async def create_tables():
    async with engine.begin() as conn:
        await conn.run_sync(metadata.create_all)


async def count_objects(session: AsyncSession, domain: type[Language] | type[User]) -> int:
    stmt = select(func.count()).select_from(domain)
    result = await session.execute(stmt)
    objects = result.scalar()
    return objects


async def main():
    await create_tables()
    english_language_domain = Language(code="en", title="English")

    # 直接传入Language对象给ui_language_code
    new_user = User(
        id=1,
        username="test",
        first_name="John",
        last_name="Doe",
        ui_language_code=english_language_domain,
    )

    async with AsyncSessionLocal() as session:
        session.add(english_language_domain)
        session.add(new_user)
        await session.commit()

        logger.info("Counting languages")
        count = await count_objects(session, domain=Language)
        assert count == 1, "Language count is not 1"
        logger.info(f"Language count is {count}")

        logger.info("Counting users")
        count = await count_objects(session, domain=User)
        assert count == 1, "User count is not 1"
        logger.info(f"User count is {count}")

        # 验证ui_language_code是Language实例且属性正确
        assert new_user.ui_language_code.code == "en"
        assert isinstance(new_user.ui_language_code, Language), f"ui_language_code is not an instance of Language, it is type of {type(new_user.ui_language_code)}"


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

关键修改说明 #

1. 映射顺序调整：先映射Language模型，确保User关联时模型已被SQLAlchemy注册。
2. 私有外键属性：_ui_language_code直接对应数据库的外键字段，业务层无需手动操作，由SQLAlchemy自动维护外键值与对象的转换。
3. 关联关系配置：将User.ui_language_code定义为relationship，指定一对一关联（uselist=False），SQLAlchemy会自动处理对象赋值时的外键存储，以及查询时的对象加载。
4. 加载策略：lazy='selectin'表示在访问ui_language_code时才查询关联对象；若需要查询User时同时加载Language，可改为lazy='joined'实现立即加载。

修改后即可直接给User的ui_language_code赋值Language对象，查询时也能通过该属性直接获取完整的Language实例，无需手动处理外键字符串。

内容的提问来源于stack exchange，提问作者Zurab Ubaev