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 文档。

或者，您可以创建具有静态属性的表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 2

从 sqlalchemy 1.1 开始，这可以使用 schema_translation_map 轻松完成。

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

Answer 3

我正在开发一个项目，在该项目中我必须动态创建 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()