sqlite 跨数据库重用ORM表

gfttwv5a  于 2023-06-23  发布在  SQLite
关注(0)|答案(1)|浏览(143)

我有一个管理多个SQLite数据库的应用程序。我尝试使用SQLAlchemy进行ORMMap,因为我尝试不重复表定义。但是,我希望一些表存在于多个数据库中。

class Mixin(MappedAsDataclass):
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, init=False, repr=False)
    
    @classmethod
    def __table_cls__(cls, table_name: str, metadata_obj: MetaData, *arg, **kwargs):
        return Table(table_name, metadata_obj, *arg, **kwargs)

class Address(Mixin):
    street: Mapped[str] = mapped_column(String)
    house_number: Mapped[int] = mapped_column(Integer)
    coordinates: Mapped[List[float]] = mapped_column(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)
    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')

这里,我想举个例子,一个数据库(我们称之为AccountDatabase)只包含Account表,另一个数据库(UserDatabase)包含所有三个表。
我正在为每个对象创建一个“数据库”对象,它应该负责Map,例如:

class AccountDatabase:
    def __init__(self, path: str, creator: Callable=None):
        self.engine = self.create_engine(path, creator)
        mapper_registry = registry()
        mapper_registry.map_imperatively(Account, Account.__table_cls__('account', mapper_registry.metadata))
        mapper_registry.metadata.create_all(self.engine)

    @staticmethod
    def create_engine(path: str, creator: Callable=None):
        """Returns a default SQLAlchemy engine for sqlite if no creator is supplied."""
        engine = create_engine(f'sqlite+pysqlite:///{path}')
        if creator is not None:
            engine = create_engine(f'sqlite+pysqlite:///{path}', creator=creator)
        return engine

然而,这似乎不起作用。我得到了错误

sqlalchemy.exc.ArgumentError: Mapper Mapper[Account(account)] could not assemble any primary key columns for mapped table 'account'

我在SQLAlchemy 2.0中尝试做的事情有可能实现吗?

jqjz2hbq

jqjz2hbq1#

在这种情况下,你应该使用一个数据库和多个模式。您可以在这里找到更多信息:https://docs.sqlalchemy.org/en/20/core/metadata.html。在这种情况下,您可以创建元数据来管理多个ORM类*,使用继承**,也可以指定table args如下:

class MyClass(Base):
    __tablename__ = "sometable"
    __table_args__ = {"schema": "some_schema"}

你可以在这里阅读更多关于它:https://docs.sqlalchemy.org/en/14/orm/declarative_tables.html

相关问题