SQLAlchemy 2.0跨数据库复用ORM表遇主键报错，能否实现？

SQLAlchemy 2.0完全支持跨数据库复用ORM表结构的需求——你可以通过为每个数据库创建独立的Registry（或MetaData）实例，将同一个ORM类映射到不同的元数据上，既复用表结构定义，又能让表存在于不同数据库中。

报错原因及修复 #

你遇到的ArgumentError: Mapper could not assemble any primary key columns错误，是因为手动调用__table_cls__创建表时，没有将Mixin中定义的id主键字段包含进去，导致映射时找不到主键。正确的做法是让SQLAlchemy自动处理字段映射，而非手动构建空表。

修复后的完整代码实现 #

1. 保留原ORM类定义

from sqlalchemy import create_engine, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import registry, relationship, Mapped, mapped_column, MappedAsDataclass
import datetime as dt
from typing import List, Callable

# 基础Mixin，定义通用主键
class Mixin(MappedAsDataclass):
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, init=False, repr=False)

# 实体类定义，复用Mixin
class Address(Mixin):
    street: Mapped[str] = mapped_column(String)
    house_number: Mapped[int] = mapped_column(Integer)
    coordinates: Mapped[List[float]] = mapped_column(ListOfFloats)  # 假设ListOfFloats为自定义类型

class Account(Mixin):
    account_id: Mapped[str] = mapped_column(String)
    balance: Mapped[float] = mapped_column(Float)

class User(Mixin):
    name: Mapped[str] = mapped_column(String)
    birthdate: Mapped[dt.datetime] = mapped_column(DateTime)
    interests: Mapped[List[str]] = mapped_column(ListOfStrings)  # 假设ListOfStrings为自定义类型
    address_id: Mapped[int] = mapped_column(Integer, ForeignKey('address.id'), init=False)
    address: Mapped[Address] = relationship(Address, foreign_keys=['address_id'], cascade='all, delete')
    account_id: Mapped[int] = mapped_column(Integer, ForeignKey('account.id'), init=False, nullable=True)
    account: Mapped[Account] = relationship(Account, foreign_keys=['account_id'], cascade='all, delete')

2. 实现仅含Account表的AccountDatabase

class AccountDatabase:
    def __init__(self, path: str, creator: Callable=None):
        self.engine = self._create_engine(path, creator)
        # 为当前数据库创建独立的Registry（包含专属MetaData）
        self.registry = registry()
        # 自动映射Account类，包含所有字段（包括Mixin的id）
        self.registry.map_imperatively(
            Account,
            include_properties=[*Account.__mapper_args__.get('properties', {}).keys(), 'id']
        )
        # 创建数据库表
        self.registry.metadata.create_all(self.engine)

    @staticmethod
    def _create_engine(path: str, creator: Callable=None):
        if creator:
            return create_engine(f'sqlite+pysqlite:///{path}', creator=creator)
        return create_engine(f'sqlite+pysqlite:///{path}')

3. 实现包含全部表的UserDatabase

class UserDatabase:
    def __init__(self, path: str, creator: Callable=None):
        self.engine = self._create_engine(path, creator)
        self.registry = registry()
        # 映射所有三个实体类到当前数据库的元数据
        for cls in [Address, Account, User]:
            self.registry.map_imperatively(
                cls,
                include_properties=[*cls.__mapper_args__.get('properties', {}).keys(), 'id']
            )
        self.registry.metadata.create_all(self.engine)

    @staticmethod
    def _create_engine(path: str, creator: Callable=None):
        if creator:
            return create_engine(f'sqlite+pysqlite:///{path}', creator=creator)
        return create_engine(f'sqlite+pysqlite:///{path}')

更简洁的声明式实现 #

如果你觉得手动调用map_imperatively繁琐，可以为每个数据库创建独立的声明式基类，动态注册实体类：

# 为不同数据库创建独立Registry和基类
account_registry = registry()
AccountBase = account_registry.generate_base(cls=MappedAsDataclass)

user_registry = registry()
UserBase = user_registry.generate_base(cls=MappedAsDataclass)

# 复用实体类定义，动态关联到对应基类
class Account(AccountBase, Mixin):
    account_id: Mapped[str] = mapped_column(String)
    balance: Mapped[float] = mapped_column(Float)

# 将同一个类注册到多个Registry（实现跨数据库复用）
user_registry.map_declaratively(Account, Mixin)
user_registry.map_declaratively(Address, Mixin)
user_registry.map_declaratively(User, Mixin)

# 简化数据库类实现
class AccountDatabase:
    def __init__(self, path: str, creator: Callable=None):
        self.engine = create_engine(f'sqlite+pysqlite:///{path}', creator=creator)
        account_registry.metadata.create_all(self.engine)

class UserDatabase:
    def __init__(self, path: str, creator: Callable=None):
        self.engine = create_engine(f'sqlite+pysqlite:///{path}', creator=creator)
        user_registry.metadata.create_all(self.engine)

关键注意事项 #

• 每个数据库必须使用独立的Registry或MetaData实例，确保表结构隔离，避免元数据冲突。
• 若数据库仅包含部分表（如AccountDatabase只有Account），不要在该数据库中映射包含跨表关系的类（如User），否则会触发外键依赖错误。
• 自定义类型（如ListOfFloats）需要确保在所有数据库中都已正确注册。

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