使用具有单个 sqlalchemy 模型的多个数据库

Posted 2023-05-08

【中文标题】使用具有单个 sqlalchemy 模型的多个数据库

【英文标题】：Using multiple databases with single sqlalchemy model

【发布时间】：2016-03-04 07:01:29

【问题描述】：

我想将多个数据库引擎与单个 sqlalchemy 数据库模型一起使用。

以下情况： 我有一个相册软件（python），不同的相册存储在不同的文件夹中。每个文件夹中都有一个单独的 sqlite 数据库，其中包含有关照片的附加信息。我不想使用单个全局数据库，因为通过这种方式，我可以简单地在文件夹基础上移动、删除和复制专辑。 打开单张专辑相当简单：

创建一个数据库会话：

maker = sessionmaker(autoflush=True, autocommit=False,
                 extension=ZopeTransactionExtension())
DBSession = scoped_session(maker)

数据库模型的基类和元数据：

DeclarativeBase = declarative_base()
metadata = DeclarativeBase.metadata

定义数据库模型（缩短）：

pic_tag_table = Table('pic_tag', metadata,
                      Column('pic_id', Integer,
                             ForeignKey('pic.pic_id'),
                             primary_key=True),
                      Column('tag_id', Integer,
                             ForeignKey('tag.tag_id'),
                             primary_key=True))


class Picture(DeclarativeBase):
    __tablename__ = 'pic'

    pic_id = Column (Integer, autoincrement = True, primary_key=True)
    ...


class Tags(DeckarativeBase):
    __tablename__ = 'tag'

    tag_id = Column (Integer, autoincrement = True, primary_key=True)
    ...

    pictures = relation('Picture', secondary=pic_tag_table, backref='tags')

最后打开连接：

engine = engine_from_config(config, '...')
DBSession.configure(bind=engine)
metadata.bind = engine

这适用于打开一张专辑。现在我想同时打开多个专辑（和数据库连接）。每张专辑都有相同的数据库模型，所以我希望我可以重用它。我的问题是模型类定义是从连接到元数据和数据库引擎的声明性基础继承的。我想将这些类连接到具有不同引擎的不同元数据。这可能吗？

P.S.：我还想通过 ORM 查询数据库，例如DBSession.query(Picture).all()（或 DBSession[0], ... 用于不同数据库上的多个会话 - 因此不是对所有数据库中的所有图片进行一次查询，而是用于查询一个数据库的 ORM 样式查询）

【参考方案1】：

您可以使用多个引擎和会话来实现这一点（您不需要多个元数据）：

engine1 = create_engine("sqlite:///tmp1.db")
engine2 = create_engine("sqlite:///tmp2.db")
Base.metadata.create_all(bind=engine1)
Base.metadata.create_all(bind=engine2)
session1 = Session(bind=engine1)
session2 = Session(bind=engine2)
print(session1.query(Picture).all())  # []
print(session2.query(Picture).all())  # []
session1.add(Picture())
session1.commit()
print(session1.query(Picture).all())  # [Picture]
print(session2.query(Picture).all())  # []
session2.add(Picture())
session2.commit()
print(session1.query(Picture).all())  # [Picture]
print(session2.query(Picture).all())  # [Picture]
session1.close()
session2.close()

对于scoped_session，您也可以创建多个。

engine1 = create_engine("sqlite:///tmp1.db")
engine2 = create_engine("sqlite:///tmp2.db")
Base.metadata.create_all(bind=engine1)
Base.metadata.create_all(bind=engine2)
Session1 = scoped_session(sessionmaker(bind=engine1))
Session2 = scoped_session(sessionmaker(bind=engine2))
session1 = Session1()
session2 = Session2()
...

如果您需要打开的数据库数量不定，scoped_session 可能会有点麻烦。您需要一些方法来跟踪它们。