运行时实体上的 sqlalchemy 动态模式

Question

我正在使用 SQL Alchemy，并且有一些特定于帐户的架构。架构的名称是使用帐户 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

一种选择是反映特定的帐户相关表。这是关于此事的SqlAlchemy Documentation。

或者，您可以使用静态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'

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

Answer 3

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

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

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)

然后在我的启动代码中：

adjust_schemas(staging=staging_name, model=model_name)

您可以将其修改为单个声明性基础。

Answer 4

我正在开发一个项目，在该项目中我必须动态创建 postgres 模式和表，然后将数据插入到适当的模式中。这是我所做的事情，也许它会对某人有所帮助：

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.models.user import User

engine_uri = "postgres://someusername:somepassword@localhost:5432/users"
engine = create_engine(engine_uri, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def create_schema(schema_name: str):
    """
        Creates a new postgres schema
        - **schema_name**: name of the new schema to create
    """
    if not engine.dialect.has_schema(engine, schema_name):
        engine.execute(sqlalchemy.schema.CreateSchema(schema_name))


def create_tables(schema_name: str):
    """
        Create new tables for postgres schema
        - **schema_name**: schema in which tables are to be created
    """
    if (
        engine.dialect.has_schema(engine, schema_name) and
        not engine.dialect.has_table(engine, str(User.__table__.name))
    ):
        User.__table__.schema = schema_name
        User.__table__.create(engine)


def add_data(schema_name: str):
    """
        Add data to a particular postgres schema
        - **schema_name**: schema in which data is to be added
    """
    if engine.dialect.has_table(engine, str(User.__table__.name)):
        db = SessionLocal()
        db.connection(execution_options={
            "schema_translate_map": {None: schema_name}},
        )
        user = User()
        user.name = "Moin"
        user.salary = 10000
        db.add(user)
        db.commit()