Postgres 模式的 SQLAlchemy 支持

Posted 2023-03-22

【中文标题】Postgres 模式的 SQLAlchemy 支持

【英文标题】：SQLAlchemy support of Postgres Schemas

【发布时间】：2012-03-07 02:01:57

【问题描述】：

我们使用 SQLAlchemy 和 postgres 托管一个多租户应用程序。我正在考虑从为每个租户拥有单独的数据库转变为具有多个模式的单个数据库。 SQLAlchemy 是否本机支持这一点？我基本上只是希望每个查询都以预先确定的模式为前缀......例如

select * from client1.users

而不仅仅是

select * from users

请注意，我想为特定请求/一组请求中的所有表切换架构，而不仅仅是这里和那里的单个表。

我想这也可以通过自定义查询类来完成，但我无法想象还没有以这种方式完成某些事情。

【参考方案1】：

嗯，有几种方法可以解决这个问题，这取决于您的应用程序的结构。这是最基本的方法：

meta = MetaData(schema="client1")

如果您的应用程序在整个应用程序中一次运行一个“客户端”，那么您就完成了。

但这里可能有问题的是，该 MetaData 中的每个表都在该架构上。如果您希望一个应用程序同时支持多个客户端（通常是“多租户”的意思），这将是笨拙的，因为您需要创建元数据的副本并复制每个客户端的所有映射。这种方法可以做到，如果你真的想这样做，它的工作方式是你可以使用特定的映射类访问每个客户端，例如：

client1_foo = Client1Foo()

在这种情况下，您将在http://www.sqlalchemy.org/trac/wiki/UsageRecipes/EntityName 与sometable.tometadata() 一起使用“实体名称”配方（请参阅http://docs.sqlalchemy.org/en/latest/core/metadata.html#sqlalchemy.schema.Table.tometadata）。

因此，假设它真正的工作方式是应用程序内有多个客户端，但每个线程一次只有一个。实际上，在 Postgresql 中最简单的方法是在开始使用连接时设置搜索路径：

# start request

# new session
sess = Session()

# set the search path
sess.execute("SET search_path TO client1")

# do stuff with session

# close it.  if you're using connection pooling, the
# search path is still set up there, so you might want to 
# revert it first
sess.close()

最后一种方法是使用 @compiles 扩展覆盖编译器，将“模式”名称粘贴到语句中。这是可行的，但会很棘手，因为生成“表”的所有地方都没有一致的挂钩。您最好的选择可能是为每个请求设置搜索路径。

【讨论】：

谢谢！我会尝试几件事，然后看看哪个效果最好并报告，但我认为路径是要走的路。

@zzzeek 我有一个镜像问题，但对于 Alembic，真的可以使用您的输入：***.com/questions/21109218/…

顺便说一句，我设法为声明性语法做到了这一点：Base = declarative_base(); Base.metadata.schema = 'ebay'。不过，可能还有更好的方法。

重要：回滚也会回滚set search_path。随后的命令将针对默认模式发出。在设置之后发出显式提交以避免这种情况。 postgresql.org/docs/current/static/sql-set.html

【参考方案2】：

如果您想在连接字符串级别执行此操作，请使用以下命令：

dbschema='schema1,schema2,public' # Searches left-to-right
engine = create_engine(
    'postgresql+psycopg2://dbuser@dbhost:5432/dbname',
    connect_args='options': '-csearch_path='.format(dbschema))

但是，对于多客户端（多租户）应用，更好的解决方案是为每个客户端配置不同的db用户，并为每个用户配置相关的search_path：

alter role user1 set search_path = "$user", public

【讨论】：

虽然这适用于 sqlalchemy，但不适用于 alembic（本机迁移工具）。如果 search_path 用于模式控制，则修订生成脚本会完全混乱。根据github.com/sqlalchemy/alembic/issues/569 的建议方法是不要触摸search_path，而是在模型和迁移中明确定义架构。

你好，这种方法适用于 mysql 吗？

【参考方案3】：

现在可以使用 Sqlalchemy 1.1 中的模式转换映射来完成。

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)

    __table_args__ = 'schema': 'per_user'

对于每个请求，可以将 Session 设置为每次引用不同的架构：

session = Session()
session.connection(execution_options=
    "schema_translate_map": "per_user": "account_one")

# will query from the ``account_one.user`` table

session.query(User).get(5)

从 SO 答案 here 中引用它。

链接到Sqlalchemy docs。

【讨论】：

如果需要在一个应用程序中使用多个模式并在类级别定义它们，这将特别有用 - 我一直在努力解决这个问题，直到找到这个答案，谢谢

父类Base和Column从何而来？

【参考方案4】：

您可以使用 sqlalchemy 事件接口来管理它。因此，在创建第一个连接之前，请按照以下方式设置一个侦听器

from sqlalchemy import event
from sqlalchemy.pool import Pool

def set_search_path( db_conn, conn_proxy ):
    print "Setting search path..."
    db_conn.cursor().execute('set search_path=client9, public')

event.listen(Pool,'connect', set_search_path )

显然这需要在创建第一个连接之前执行（例如在应用程序初始化中）

我看到的 session.execute(...) 解决方案的问题是它在会话使用的特定连接上执行。但是，我在 sqlalchemy 中看不到任何保证会话将无限期地继续使用相同连接的东西。如果它从连接池中拾取一个新的连接，那么它将丢失搜索路径设置。

我需要这样的方法来设置应用程序 search_path，这与数据库或用户搜索路径不同。我希望能够在引擎配置中进行设置，但看不到这样做的方法。使用连接事件确实有效。如果有人有，我会对更简单的解决方案感兴趣。

另一方面，如果您想在一个应用程序中处理多个客户端，那么这将不起作用 - 我猜 session.execute(...) 方法可能是最好的方法。

【讨论】：

您是否有一种优雅的方式将“client9”作为参数传递而不是硬编码？我目前（hacky）的解决方法是将application_name 查询参数传递给db-url（?application_name=bla），然后在set_search_path 和db_conn.dsn.split('application_name=')[1]) 中检查它。

【参考方案5】：

Table definitions 中有一个架构属性

我不确定它是否有效，但您可以尝试：

Table(CP.get('users', metadata, schema='client1',....)

【讨论】：

我正在寻找更全球化的东西，以便我可以为单个请求切换所有表中的所有查询。我将更新问题以反映这一点。

【参考方案6】：

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

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

connection = engine.connect().execution_options(
    schema_translate_map=None: "user_schema_one")

result = connection.execute(user_table.select())

以下是所有可用选项的详细评论： https://github.com/sqlalchemy/sqlalchemy/issues/4081

【参考方案7】：

我试过了：

con.execute('SET search_path TO schema'.format(schema='myschema'))

这对我不起作用。然后我在 init 函数中使用了 schema= 参数：

# We then bind the connection to MetaData()
meta = sqlalchemy.MetaData(bind=con, reflect=True, schema='myschema')

然后我用架构名称限定表

house_table = meta.tables['myschema.houses']

一切正常。

【参考方案8】：

可以在数据库级别解决此问题。我想您的应用程序有一个专用用户，该用户被授予架构上的一些特权。只需为他设置search_path 到这个模式：

ALTER ROLE your_user IN DATABASE your_db SET search_path TO your_schema;

【参考方案9】：

您可以更改您的 search_path。问题

set search_path=client9;

在您的会话开始时，然后保持您的表不合格。

您还可以在每个数据库或每个用户级别设置默认搜索路径。默认情况下，我很想将其设置为空架构，这样您就可以轻松发现任何设置失败。

http://www.postgresql.org/docs/current/static/ddl-schemas.html#DDL-SCHEMAS-PATH

【讨论】：

那是……一个好主意。高五！

请记住，在 session.commit() 之后会启动一个新事务，因此 search_path 将被重置。 SA 会话事件非常适合为每个新事务设置 search_path。

【参考方案10】：

我发现上述答案均不适用于 SqlAlchmeny 1.2.4。这是对我有用的解决方案。

from sqlalchemy import MetaData, Table
from sqlalchemy import create_engine    

def table_schemato_psql(schema_name, table_name):

        conn_str = 'postgresql://username:password@localhost:5432/database'.format(
            username='<username>',
            password='<password>',
            database='<database name>'
        )

        engine = create_engine(conn_str)

        with engine.connect() as conn:
            conn.execute('SET search_path TO schema'.format(schema=schema_name))

            meta = MetaData()

            table_data = Table(table_name, meta,
                              autoload=True,
                              autoload_with=conn,
                              postgresql_ignore_search_path=True)

            for column in table_data.columns:
                print column.name

【参考方案11】：

我使用以下模式。

engine = sqlalchemy.create_engine("postgresql://postgres:mypass@172.17.0.2/mydb")

for schema in ['schema1', 'schema2']:
    engine.execute(CreateSchema(schema))
    tmp_engine = engine.execution_options(schema_translate_map =  None: schema  )
    Base.metadata.create_all(tmp_engine)

【参考方案12】：

各位过来人，更通用的可以支持MYSQL或Oracle的解决方案，请参考this guide。

所以基本上它会在第一次连接到数据库时为引擎设置架构。

engine = create_engine("engine_url")

@event.listens_for(engine, "connect", insert=True)
def set_current_schema(dbapi_connection, connection_record):
    cursor_obj = dbapi_connection.cursor()
    cursor_obj.execute(f"USE self.schemas_name")
    cursor_obj.close()

要执行的查询取决于您使用的数据库，因此对于 PSQL，您将有不同的查询，对于 ORACLE，您将有不同的查询，等等。