运行时实体上的sqlalchemy动态架构

Question

我使用的是SQL炼金术，并且有一些特定于帐户的模式。模式的名称是使用帐户ID派生的，因此，在找到应用程序服务或存储库层之前，我没有模式的名称。我想知道是否可以对运行时动态设置模式的实体运行查询？在

我知道我需要设置__table_args__['schema']，并尝试使用内置的type（）进行设置，但始终会出现以下错误：

could not assemble any primary key columns for mapped table

我已经准备好放弃直接写sql了，但我真的很讨厌这样做。你知道怎么做吗？我用的是SA 0.99，我有一个PK图。在

谢谢

Answer 1

从sqlalchemy 1.1中， 这可以通过使用schema_translation_map轻松完成。在

https://docs.sqlalchemy.org/en/11/changelog/migration_11.html#multi-tenancy-schema-translation-for-table-objects

Answer 2

编辑：虽然您可以做我在这里所做的，但是使用下面的答案所示的模式转换映射是正确的方法。在

它们是静态设置的。外键需要同样的处理方式，而且我还有一个额外的问题，因为我有多个模式包含多个表，所以我这样做了：

from sqlalchemy.ext.declarative import declarative_base

staging_dbase = declarative_base()
model_dbase = declarative_base()


def adjust_schemas(staging, model):
    for vv in staging_dbase.metadata.tables.values():
        vv.schema = staging
    for vv in model_dbase.metadata.tables.values():
        vv.schema = model


def all_tables():
    return staging_dbase.metadata.tables.union(model_dbase.metadata.tables)

然后在我的启动代码中：

^{pr2}$

您可以为一个声明性基修改它。在

Answer 3

一种选择是反映特定的账户依赖表。这是关于这件事的SqlAlchemy Documentation。在

或者，您可以使用静态schema属性创建表，并在运行时根据需要对其进行更新，并运行所需的查询。我想不出一个不乱的方法来做这件事。所以这是一个混乱的选择

每当切换帐户时，使用循环更新每个表定义中的schema属性。在

1. 将帐户特定的所有表添加到列表中。

2. • 如果表是用声明性语法表示的，那么您必须修改DeclarativeName.__table__.schema属性。我不确定您是否还需要修改DeclarativeName.__table_args__['schema']，但我想这不会有什么坏处。在
   • 如果表是用旧样式的表语法表示的，则必须修改Table.schema属性。在

3. 如果您将文本用于任何关系或外键，那么这将中断，您必须检查每个表是否存在这种硬编码用法并更改它们
   示例

   • user_id = Column(ForeignKey('my_schema.user.id'))需要写成user_id = Column(ForeignKey(User.id))。然后您可以将User的模式更改为my_new_schema。否则，在查询时sqlalchemy将被混淆，因为外键将指向my_schema.user.id，而查询将指向my_new_schema.user。在

4. 我不确定是否可以在不使用纯文本的情况下表达更复杂的关系，所以我想这是我提出的解决方案的限制。

下面是我在终端上写的一个例子：

>>> from sqlalchemy import Column, Table, Integer, String, select, ForeignKey
>>> from sqlalchemy.orm import relationship, backref
>>> from sqlalchemy.ext.declarative import declarative_base
>>> B = declarative_base()
>>> 
>>> class User(B):
...   __tablename__ = 'user'
...   __table_args__ = {'schema': 'first_schema'}
...   id = Column(Integer, primary_key=True)
...   name = Column(String)
...   email = Column(String)
... 
>>> class Posts(B):
...   __tablename__ = 'posts'
...   __table_args__ = {'schema':'first_schema'}
...   id = Column(Integer, primary_key=True)
...   user_id = Column(ForeignKey(User.id))
...   text = Column(String)
... 
>>> str(select([User.id, Posts.text]).select_from(User.__table__.join(Posts)))
'SELECT first_schema."user".id, first_schema.posts.text \nFROM first_schema."user" JOIN first_schema.posts ON first_schema."user".id = first_schema.posts.user_id'
>>> account_specific = [User, Posts]
>>> for Tbl in account_specific:
...   Tbl.__table__.schema = 'second_schema'
... 
>>> str(select([User.id, Posts.text]).select_from(User.__table__.join(Posts)))
'SELECT second_schema."user".id, second_schema.posts.text \nFROM second_schema."user" JOIN second_schema.posts ON second_schema."user".id = second_schema.posts.user_id'

如您所见，在我更新表的schema属性之后，相同的查询引用了second_schema。在