SqlAlchemy 2.0 中文文档(四十八)(2)https://developer.aliyun.com/article/1563008
事务隔离级别
大多数 SQLAlchemy 方言支持使用create_engine.isolation_level
参数在create_engine()
级别和在Connection
级别通过Connection.execution_options.isolation_level
参数设置事务隔离级别。
对于 PostgreSQL 方言,此功能通过利用 DBAPI 特定功能实现,例如 psycopg2 的隔离级别标志,该标志将隔离级别设置嵌入到与"BEGIN"
语句一起发出的语句中,或者对于没有直接支持的 DBAPI,通过在 DBAPI 发出的"BEGIN"
语句之前发出SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL
来实现。对于特殊的 AUTOCOMMIT 隔离级别,使用 DBAPI 特定的技术,通常是 DBAPI 连接对象上的.autocommit
标志。
要使用create_engine()
设置隔离级别:
engine = create_engine( "postgresql+pg8000://scott:tiger@localhost/test", isolation_level = "REPEATABLE READ" )
要使用每个连接的执行选项设置:
with engine.connect() as conn: conn = conn.execution_options( isolation_level="REPEATABLE READ" ) with conn.begin(): # ... work with transaction
还有更多关于隔离级别配置的选项,例如与主Engine
关联的“子引擎”对象,每个对象应用不同的隔离级别设置。有关背景信息,请参阅设置事务隔离级别,包括 DBAPI 自动提交中的讨论。
大多数 PostgreSQL 方言的isolation_level
的有效值包括:
READ COMMITTED
READ UNCOMMITTED
REPEATABLE READ
SERIALIZABLE
AUTOCOMMIT
另请参见
设置事务隔离级别,包括 DBAPI 自动提交
设置只读 / 可延迟
Psycopg2 事务隔离级别
pg8000 事务隔离级别
设置只读 / 可延迟
大多数 PostgreSQL 方言支持设置事务的“只读”和“可延迟”特性,这是隔离级别设置的补充。这两个属性可以通过使用Connection.execution_options()
传递postgresql_readonly
和postgresql_deferrable
标志来同时或独立地建立。下面的示例演示了同时设置“只读”和“可延迟”以及设置"SERIALIZABLE"
隔离级别:
with engine.connect() as conn: conn = conn.execution_options( isolation_level="SERIALIZABLE", postgresql_readonly=True, postgresql_deferrable=True ) with conn.begin(): # ... work with transaction
请注意,一些 DBAPI(如 asyncpg)仅支持“只读”与 SERIALIZABLE 隔离。
新版本 1.4 中:增加了对postgresql_readonly
和postgresql_deferrable
执行选项的支持。
用于连接池的临时表 / 资源重置
SQLAlchemy Engine
对象使用的 QueuePool
连接池实现包括在连接返回到池时调用 DBAPI 的 .rollback()
方法的 reset on return 行为。虽然这个回滚会清除前一个事务使用的即时状态,但它不涵盖更广泛的会话级状态,包括临时表以及其他服务器状态,如预备声明句柄和语句缓存。PostgreSQL 数据库包括各种命令,可用于重置此状态,包括 DISCARD
、RESET
、DEALLOCATE
和 UNLISTEN
。
要将其中一个或多个命令安装为执行返回时的重置手段,可以使用 PoolEvents.reset()
事件钩子,如下面的示例所示。该实现将结束进行中的事务,并使用 CLOSE
、RESET
和 DISCARD
命令丢弃临时表;有关每个语句的背景,请参阅 PostgreSQL 文档。
create_engine.pool_reset_on_return
参数设置为 None
,以便自定义方案可以完全替换默认行为。自定义钩子实现在任何情况下都调用 .rollback()
,因为通常重要的是 DBAPI 自身的提交/回滚跟踪将与事务的状态保持一致:
from sqlalchemy import create_engine from sqlalchemy import event postgresql_engine = create_engine( "postgresql+pyscopg2://scott:tiger@hostname/dbname", # disable default reset-on-return scheme pool_reset_on_return=None, ) @event.listens_for(postgresql_engine, "reset") def _reset_postgresql(dbapi_connection, connection_record, reset_state): if not reset_state.terminate_only: dbapi_connection.execute("CLOSE ALL") dbapi_connection.execute("RESET ALL") dbapi_connection.execute("DISCARD TEMP") # so that the DBAPI itself knows that the connection has been # reset dbapi_connection.rollback()
从版本 2.0.0b3 起更改:为 PoolEvents.reset()
事件添加了额外的状态参数,并确保事件对所有“重置”事件都会被调用,因此它适用于自定义“重置”处理程序的位置。之前使用 PoolEvents.checkin()
处理程序的方案仍然可用。
另请参阅
返回时重置 - 在连接池文档中
在连接时设置备用搜索路径
PostgreSQL 的 search_path
变量是指在引用特定表或其他对象时将隐式引用的模式名称列表。如下一节远程模式表内省和 PostgreSQL search_path 所述,SQLAlchemy 通常以保持此变量处于其默认值public
的状态为组织方式,但是,为了在自动使用连接时将其设置为任意名称或名称,可以使用以下事件处理程序为池中的所有连接调用“SET SESSION search_path”命令,如设置新连接的默认模式所讨论的那样:
from sqlalchemy import event from sqlalchemy import create_engine engine = create_engine("postgresql+psycopg2://scott:tiger@host/dbname") @event.listens_for(engine, "connect", insert=True) def set_search_path(dbapi_connection, connection_record): existing_autocommit = dbapi_connection.autocommit dbapi_connection.autocommit = True cursor = dbapi_connection.cursor() cursor.execute("SET SESSION search_path='%s'" % schema_name) cursor.close() dbapi_connection.autocommit = existing_autocommit
由于使用了 .autocommit
DBAPI 属性,该命令的复杂之处在于当调用SET SESSION search_path
指令时,它是在任何事务范围之外调用的,因此当 DBAPI 连接回滚时不会被还原。
另请参阅
设置新连接的默认模式 - 在 用 MetaData 描述数据库 文档中
远程模式表内省和 PostgreSQL search_path
最佳实践摘要部分
保持search_path
变量设置为其默认值public
,不带任何其他模式名称。确保用于连接的用户名不与远程模式匹配,或者确保从search_path
中移除"$user"
标记。对于其他模式名称,请在Table
定义中明确命名这些。另外,postgresql_ignore_search_path
选项将导致所有反映的Table
对象设置一个Table.schema
属性。
PostgreSQL 方言可以反映来自任何模式的表,如 从其他模式反射表 中所述。
在所有情况下,SQLAlchemy 反射表时的第一件事是确定当前数据库连接的默认模式。它使用 PostgreSQL 的 current_schema()
函数来执行此操作,下面使用 PostgreSQL 客户端会话(即使用 psql
工具)进行演示:
test=> select current_schema(); current_schema ---------------- public (1 row)
如上所述,在普通安装的 PostgreSQL 上,默认模式名称是名称public
。
然而,如果您的数据库用户名与模式名称匹配,PostgreSQL 的默认行为是将该名称用作默认模式。下面,我们使用用户名scott
登录。当我们创建一个名为scott
的模式时,它会隐式地更改默认模式:
test=> select current_schema(); current_schema ---------------- public (1 row) test=> create schema scott; CREATE SCHEMA test=> select current_schema(); current_schema ---------------- scott (1 row)
current_schema()
函数的行为源自 PostgreSQL 搜索路径 变量 search_path
,在现代 PostgreSQL 版本中,默认情况如下:
test=> show search_path; search_path ----------------- "$user", public (1 row)
在上述情况下,"$user"
变量将注入当前用户名作为默认模式,如果存在的话。否则,将使用 public
。
当 Table
对象反映时,如果它存在于由 current_schema()
函数指示的模式中,表的“.schema”属性分配的模式名称是 Python 的“None”值。否则,“.schema”属性将被分配为该模式的字符串名称。
关于这些 Table
对象通过外键约束引用的表,必须决定在当前 search_path
中的远程模式名称在那些远程表中如何表示。
默认情况下,PostgreSQL 方言模仿 PostgreSQL 自己的 pg_get_constraintdef()
内置过程鼓励的行为。当引用的模式名称也在 PostgreSQL 模式搜索路径中时,此函数从该定义中省略引用模式名称,该定义返回特定外键约束的示例定义。下面的交互演示了此行为:
test=> CREATE TABLE test_schema.referred(id INTEGER PRIMARY KEY); CREATE TABLE test=> CREATE TABLE referring( test(> id INTEGER PRIMARY KEY, test(> referred_id INTEGER REFERENCES test_schema.referred(id)); CREATE TABLE test=> SET search_path TO public, test_schema; test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n test-> ON n.oid = c.relnamespace test-> JOIN pg_catalog.pg_constraint r ON c.oid = r.conrelid test-> WHERE c.relname='referring' AND r.contype = 'f' test-> ; pg_get_constraintdef --------------------------------------------------- FOREIGN KEY (referred_id) REFERENCES referred(id) (1 row)
在上面,我们创建了一个名为 referred
的表,作为远程模式 test_schema
的成员,然而当我们将 test_schema
添加到 PG 的 search_path
然后询问 pg_get_constraintdef()
关于 FOREIGN KEY
语法时,test_schema
没有包含在函数的输出中。
另一方面,如果我们将搜索路径设置回典型默认的 public
:
test=> SET search_path TO public; SET
现在相同的查询对于 pg_get_constraintdef()
返回了完全限定的名称:
test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n test-> ON n.oid = c.relnamespace test-> JOIN pg_catalog.pg_constraint r ON c.oid = r.conrelid test-> WHERE c.relname='referring' AND r.contype = 'f'; pg_get_constraintdef --------------------------------------------------------------- FOREIGN KEY (referred_id) REFERENCES test_schema.referred(id) (1 row)
SQLAlchemy 默认使用 pg_get_constraintdef()
的返回值来确定远程模式名称。也就是说,如果我们的 search_path
被设置为包含 test_schema
,并且我们调用了如下表反射过程:
>>> from sqlalchemy import Table, MetaData, create_engine, text >>> engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test") >>> with engine.connect() as conn: ... conn.execute(text("SET search_path TO test_schema, public")) ... metadata_obj = MetaData() ... referring = Table('referring', metadata_obj, ... autoload_with=conn) ... <sqlalchemy.engine.result.CursorResult object at 0x101612ed0>
上述过程将向 MetaData.tables
集合交付 不含 模式的 referred
表名称:
>>> metadata_obj.tables['referred'].schema is None True
要修改反射行为,使参考模式不受 search_path
设置的影响,请使用 postgresql_ignore_search_path
选项,该选项可以指定为 Table
和 MetaData.reflect()
的方言特定参数:
>>> with engine.connect() as conn: ... conn.execute(text("SET search_path TO test_schema, public")) ... metadata_obj = MetaData() ... referring = Table('referring', metadata_obj, ... autoload_with=conn, ... postgresql_ignore_search_path=True) ... <sqlalchemy.engine.result.CursorResult object at 0x1016126d0>
现在,我们将 test_schema.referred
作为模式限定存储:
>>> metadata_obj.tables['test_schema.referred'].schema 'test_schema'
另请参阅
模式限定反射与默认模式的交互 - 从与后端无关的角度讨论这个问题
模式搜索路径 - 在 PostgreSQL 网站上。
INSERT/UPDATE…RETURNING
该方言支持 PG 8.2 的 INSERT..RETURNING
、UPDATE..RETURNING
和 DELETE..RETURNING
语法。默认情况下,对于单行 INSERT 语句,使用 INSERT..RETURNING
来获取新生成的主键标识符。要在每个语句基础上指定显式的 RETURNING
子句,请使用 _UpdateBase.returning()
方法:
# INSERT..RETURNING result = table.insert().returning(table.c.col1, table.c.col2).\ values(name='foo') print(result.fetchall()) # UPDATE..RETURNING result = table.update().returning(table.c.col1, table.c.col2).\ where(table.c.name=='foo').values(name='bar') print(result.fetchall()) # DELETE..RETURNING result = table.delete().returning(table.c.col1, table.c.col2).\ where(table.c.name=='foo') print(result.fetchall())
INSERT…ON CONFLICT(插入或更新)
从版本 9.5 开始,PostgreSQL 允许通过 INSERT
语句的 ON CONFLICT
子句将行“插入或更新”到表中。只有在该行不违反任何唯一约束的情况下才会插入候选行。在唯一约束冲突的情况下,可以发生次要操作,可以是“DO UPDATE”,表示应更新目标行中的数据,或者是“DO NOTHING”,表示静默跳过此行。
冲突是使用现有的唯一约束和索引确定的。这些约束可以通过在 DDL 中声明的名称来标识,也可以通过声明组成索引的列和条件来推断。
SQLAlchemy 通过 PostgreSQL 特定的 insert()
函数提供 ON CONFLICT
支持,该函数提供了生成方法 Insert.on_conflict_do_update()
和 Insert.on_conflict_do_nothing()
:
>>> from sqlalchemy.dialects.postgresql import insert >>> insert_stmt = insert(my_table).values( ... id='some_existing_id', ... data='inserted value') >>> do_nothing_stmt = insert_stmt.on_conflict_do_nothing( ... index_elements=['id'] ... ) >>> print(do_nothing_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO NOTHING >>> do_update_stmt = insert_stmt.on_conflict_do_update( ... constraint='pk_my_table', ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT ON CONSTRAINT pk_my_table DO UPDATE SET data = %(param_1)s
另请参阅
INSERT … ON CONFLICT - 在 PostgreSQL 文档中。
指定目标
这两种方法都使用命名约束或通过列推断来提供冲突的“目标”:
Insert.on_conflict_do_update.index_elements
参数指定了一个序列,其中包含字符串列名、Column
对象和/或 SQL 表达式元素,这些元素将标识一个唯一索引:
>>> do_update_stmt = insert_stmt.on_conflict_do_update( ... index_elements=['id'], ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s >>> do_update_stmt = insert_stmt.on_conflict_do_update( ... index_elements=[my_table.c.id], ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
- 当使用
Insert.on_conflict_do_update.index_elements
推断索引时,还可以通过指定Insert.on_conflict_do_update.index_where
参数来推断部分索引:
>>> stmt = insert(my_table).values(user_email='a@b.com', data='inserted data') >>> stmt = stmt.on_conflict_do_update( ... index_elements=[my_table.c.user_email], ... index_where=my_table.c.user_email.like('%@gmail.com'), ... set_=dict(data=stmt.excluded.data) ... ) >>> print(stmt) INSERT INTO my_table (data, user_email) VALUES (%(data)s, %(user_email)s) ON CONFLICT (user_email) WHERE user_email LIKE %(user_email_1)s DO UPDATE SET data = excluded.data
Insert.on_conflict_do_update.constraint
参数用于直接指定索引而不是推断索引。这可以是唯一约束的名称、主键约束或索引:
>>> do_update_stmt = insert_stmt.on_conflict_do_update( ... constraint='my_table_idx_1', ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT ON CONSTRAINT my_table_idx_1 DO UPDATE SET data = %(param_1)s >>> do_update_stmt = insert_stmt.on_conflict_do_update( ... constraint='my_table_pk', ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT ON CONSTRAINT my_table_pk DO UPDATE SET data = %(param_1)s
Insert.on_conflict_do_update.constraint
参数也可以引用表示约束的 SQLAlchemy 构造,例如UniqueConstraint
、PrimaryKeyConstraint
、Index
或ExcludeConstraint
。在这种用法中,如果约束有名称,则直接使用。否则,如果约束没有名称,则将使用推断,其中约束的表达式和可选的 WHERE 子句将在构造中详细说明。这种用法特别方便,可以使用Table.primary_key
属性来引用具有命名或未命名主键的Table
:
>>> do_update_stmt = insert_stmt.on_conflict_do_update( ... constraint=my_table.primary_key, ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
SET 子句
ON CONFLICT...DO UPDATE
用于执行已存在行的更新,可以使用新值和插入提议中的值的任意组合。这些值使用 Insert.on_conflict_do_update.set_
参数指定。该参数接受一个包含直接更新值的字典:
>>> stmt = insert(my_table).values(id='some_id', data='inserted value') >>> do_update_stmt = stmt.on_conflict_do_update( ... index_elements=['id'], ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
警告
Insert.on_conflict_do_update()
方法不会考虑 Python 端的默认 UPDATE 值或生成函数,例如使用Column.onupdate
指定的值。除非在Insert.on_conflict_do_update.set_
字典中手动指定,否则这些值不会用于 ON CONFLICT 风格的 UPDATE。
使用被排除的插入值进行更新
为了引用提议的插入行,特殊别名Insert.excluded
可作为Insert
对象的属性使用;此对象是一��包含目标表的所有列的ColumnCollection
别名:
>>> stmt = insert(my_table).values( ... id='some_id', ... data='inserted value', ... author='jlh' ... ) >>> do_update_stmt = stmt.on_conflict_do_update( ... index_elements=['id'], ... set_=dict(data='updated value', author=stmt.excluded.author) ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data, author) VALUES (%(id)s, %(data)s, %(author)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author
附加的 WHERE 条件
Insert.on_conflict_do_update()
方法还接受使用Insert.on_conflict_do_update.where
参数的 WHERE 子句,这将限制接收 UPDATE 的行:
>>> stmt = insert(my_table).values( ... id='some_id', ... data='inserted value', ... author='jlh' ... ) >>> on_update_stmt = stmt.on_conflict_do_update( ... index_elements=['id'], ... set_=dict(data='updated value', author=stmt.excluded.author), ... where=(my_table.c.status == 2) ... ) >>> print(on_update_stmt) INSERT INTO my_table (id, data, author) VALUES (%(id)s, %(data)s, %(author)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author WHERE my_table.status = %(status_1)s
使用 DO NOTHING 跳过行
ON CONFLICT
可用于在出现与唯一性或排他性约束冲突时完全跳过插入行;下面通过Insert.on_conflict_do_nothing()
方法进行了说明:
>>> stmt = insert(my_table).values(id='some_id', data='inserted value') >>> stmt = stmt.on_conflict_do_nothing(index_elements=['id']) >>> print(stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO NOTHING
如果使用DO NOTHING
而没有指定任何列或约束条件,则会跳过任何唯一性或排他性约束违规的插入操作:
>>> stmt = insert(my_table).values(id='some_id', data='inserted value') >>> stmt = stmt.on_conflict_do_nothing() >>> print(stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT DO NOTHING
指定目标
两种方法都使用命名约束或通过列推断来提供冲突的“目标”:
Insert.on_conflict_do_update.index_elements
参数指定了一个包含字符串列名、Column
对象和/或 SQL 表达式元素的序列,用于标识唯一索引:
>>> do_update_stmt = insert_stmt.on_conflict_do_update( ... index_elements=['id'], ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s >>> do_update_stmt = insert_stmt.on_conflict_do_update( ... index_elements=[my_table.c.id], ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
- 当使用
Insert.on_conflict_do_update.index_elements
推断索引时,也可以通过同时指定使用Insert.on_conflict_do_update.index_where
参数来推断部分索引:
>>> stmt = insert(my_table).values(user_email='a@b.com', data='inserted data') >>> stmt = stmt.on_conflict_do_update( ... index_elements=[my_table.c.user_email], ... index_where=my_table.c.user_email.like('%@gmail.com'), ... set_=dict(data=stmt.excluded.data) ... ) >>> print(stmt) INSERT INTO my_table (data, user_email) VALUES (%(data)s, %(user_email)s) ON CONFLICT (user_email) WHERE user_email LIKE %(user_email_1)s DO UPDATE SET data = excluded.data
Insert.on_conflict_do_update.constraint
参数用于直接指定索引,而不是推断索引。这可以是唯一约束、主键约束或索引的名称:
>>> do_update_stmt = insert_stmt.on_conflict_do_update( ... constraint='my_table_idx_1', ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT ON CONSTRAINT my_table_idx_1 DO UPDATE SET data = %(param_1)s >>> do_update_stmt = insert_stmt.on_conflict_do_update( ... constraint='my_table_pk', ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT ON CONSTRAINT my_table_pk DO UPDATE SET data = %(param_1)s
Insert.on_conflict_do_update.constraint
参数也可以指代一个 SQLAlchemy 构造,代表一个约束,例如UniqueConstraint
、PrimaryKeyConstraint
、Index
或ExcludeConstraint
。在这种用法中,如果约束有名称,则直接使用它。否则,如果约束没有名称,则将使用推断,其中约束的表达式和可选的 WHERE 子句将在构造中详细说明。这种用法特别方便,可以使用Table.primary_key
属性引用表的命名或未命名的主键:
>>> do_update_stmt = insert_stmt.on_conflict_do_update( ... constraint=my_table.primary_key, ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
SET 子句
ON CONFLICT...DO UPDATE
用于执行已经存在的行的更新,使用新值以及来自建议插入的值的任意组合。这些值使用 Insert.on_conflict_do_update.set_
参数指定。此参数接受一个包含直接更新值的字典:
>>> stmt = insert(my_table).values(id='some_id', data='inserted value') >>> do_update_stmt = stmt.on_conflict_do_update( ... index_elements=['id'], ... set_=dict(data='updated value') ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
警告
Insert.on_conflict_do_update()
方法不会考虑 Python 端的默认更新值或生成函数,例如使用 Column.onupdate
指定的值。这些值不会在 ON CONFLICT 类型的更新中执行,除非它们在 Insert.on_conflict_do_update.set_
字典中手动指定。
使用排除的插入值进行更新
要引用提议的插入行,Insert.excluded
这个特殊别名可作为 Insert
对象的属性;这个对象是一个包含目标表所有列的 ColumnCollection
别名:
>>> stmt = insert(my_table).values( ... id='some_id', ... data='inserted value', ... author='jlh' ... ) >>> do_update_stmt = stmt.on_conflict_do_update( ... index_elements=['id'], ... set_=dict(data='updated value', author=stmt.excluded.author) ... ) >>> print(do_update_stmt) INSERT INTO my_table (id, data, author) VALUES (%(id)s, %(data)s, %(author)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author
附加的 WHERE 条件
Insert.on_conflict_do_update()
方法还接受一个 WHERE 子句,使用 Insert.on_conflict_do_update.where
参数,将限制那些接收更新的行:
>>> stmt = insert(my_table).values( ... id='some_id', ... data='inserted value', ... author='jlh' ... ) >>> on_update_stmt = stmt.on_conflict_do_update( ... index_elements=['id'], ... set_=dict(data='updated value', author=stmt.excluded.author), ... where=(my_table.c.status == 2) ... ) >>> print(on_update_stmt) INSERT INTO my_table (id, data, author) VALUES (%(id)s, %(data)s, %(author)s) ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author WHERE my_table.status = %(status_1)s
使用 DO NOTHING 跳过行
ON CONFLICT
可用于完全跳过插入行,如果与唯一约束或排除约束发生冲突;下面是使用 Insert.on_conflict_do_nothing()
方法进行说明:
>>> stmt = insert(my_table).values(id='some_id', data='inserted value') >>> stmt = stmt.on_conflict_do_nothing(index_elements=['id']) >>> print(stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT (id) DO NOTHING
如果使用 DO NOTHING
而没有指定任何列或约束,它将跳过任何唯一或排除约束违规的插入:
>>> stmt = insert(my_table).values(id='some_id', data='inserted value') >>> stmt = stmt.on_conflict_do_nothing() >>> print(stmt) INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s) ON CONFLICT DO NOTHING
全文搜索
PostgreSQL 的全文搜索系统可通过 func
命名空间使用,结合通过 Operators.bool_op()
方法使用自定义运算符。对于一些简单情况,具有一定跨后端兼容性,也可以使用 Operators.match()
运算符。
使用 match()
进行简单纯文本���配
Operators.match()
运算符提供了跨平台的简单文本匹配功能。对于 PostgreSQL 后端,它被硬编码为使用 @@
运算符与 plainto_tsquery()
PostgreSQL 函数结合生成表达式。
在 PostgreSQL 方言中,类似以下表达式:
select(sometable.c.text.match("search string"))
将发送到数据库的内容:
SELECT text @@ plainto_tsquery('search string') FROM table
在上面的示例中,将普通字符串传递给Operators.match()
将自动使用plainto_tsquery()
来指定 tsquery 的类型。这为Operators.match()
与其他后端的基本数据库跨兼容性建立了基础。
从版本 2.0 开始更改:PostgreSQL 方言与Operators.match()
一起使用的默认 tsquery 生成函数是plainto_tsquery()
。
要精确呈现 1.4 中呈现的内容,请使用以下形式:
from sqlalchemy import func select( sometable.c.text.bool_op("@@")(func.to_tsquery("search string")) )
这将生成:
SELECT text @@ to_tsquery('search string') FROM table
直接使用 PostgreSQL 全文函数和运算符
文本搜索操作超出了简单使用Operators.match()
,可能会利用func
命名空间生成 PostgreSQL 全文函数,结合Operators.bool_op()
生成任何布尔运算符。
例如,查询:
select( func.to_tsquery('cat').bool_op("@>")(func.to_tsquery('cat & rat')) )
将生成:
SELECT to_tsquery('cat') @> to_tsquery('cat & rat')
TSVECTOR
类型可以提供显式 CAST:
from sqlalchemy.dialects.postgresql import TSVECTOR from sqlalchemy import select, cast select(cast("some text", TSVECTOR))
生成等效于的语句:
SELECT CAST('some text' AS TSVECTOR) AS anon_1
PostgreSQL 方言通过增强func
命名空间来设置大多数全文搜索函数的正确参数和返回类型。假定已导入sqlalchemy.dialects.postgresql
包,或者使用postgresql
方言调用了create_engine()
,这些函数将自动使用sqlalchemy.sql.expression.func
命名空间。这些函数在以下文档中有详细说明:
to_tsvector
to_tsquery
plainto_tsquery
phraseto_tsquery
websearch_to_tsquery
ts_headline
使用match()
或自定义运算符指定“regconfig”
PostgreSQL 的plainto_tsquery()
函数接受一个可选的“regconfig”参数,用于指示 PostgreSQL 使用特定的预计算 GIN 或 GiST 索引来执行搜索。在使用Operators.match()
时,可以使用postgresql_regconfig
参数指定此附加参数,例如:
select(mytable.c.id).where( mytable.c.title.match('somestring', postgresql_regconfig='english') )
这将生成:
SELECT mytable.id FROM mytable WHERE mytable.title @@ plainto_tsquery('english', 'somestring')
当使用其他 PostgreSQL 搜索函数与 func
时,“regconfig” 参数可以直接作为初始参数传递:
select(mytable.c.id).where( func.to_tsvector("english", mytable.c.title).bool_op("@@")( func.to_tsquery("english", "somestring") ) )
产生一个等效于的语句:
SELECT mytable.id FROM mytable WHERE to_tsvector('english', mytable.title) @@ to_tsquery('english', 'somestring')
建议使用 PostgreSQL 的 EXPLAIN ANALYZE...
工具,以确保您使用 SQLAlchemy 生成了充分利用为全文搜索创建的任何索引的查询。
另请参阅
全文搜索 - PostgreSQL 文档中
使用 match()
进行简单纯文本匹配
Operators.match()
运算符提供跨兼容的简单文本匹配。对于 PostgreSQL 后端,它被硬编码为使用 @@
运算符与 plainto_tsquery()
PostgreSQL 函数结合生成表达式。
在 PostgreSQL 方言上,类似以下表达式:
select(sometable.c.text.match("search string"))
将会发送到数据库:
SELECT text @@ plainto_tsquery('search string') FROM table
在上面,将纯字符串传递给 Operators.match()
将自动使用 plainto_tsquery()
来指定 tsquery 的类型。这为 Operators.match()
与其他后端的基本数据库跨兼容性��立了基础。
从版本 2.0 开始更改:PostgreSQL 方言与 Operators.match()
一起使用的默认 tsquery 生成函数是 plainto_tsquery()
。
要精确呈现 1.4 中呈现的内容,请使用以下形式:
from sqlalchemy import func select( sometable.c.text.bool_op("@@")(func.to_tsquery("search string")) )
这将生成:
SELECT text @@ to_tsquery('search string') FROM table
直接使用 PostgreSQL 全文搜索函数和运算符
超出简单使用 Operators.match()
的文本搜索操作可能会利用 func
命名空间来生成 PostgreSQL 全文搜索函数,结合 Operators.bool_op()
生成任何布尔运算符。
例如,查询:
select( func.to_tsquery('cat').bool_op("@>")(func.to_tsquery('cat & rat')) )
会生成:
SELECT to_tsquery('cat') @> to_tsquery('cat & rat')
TSVECTOR
类型可以提供显式转换:
from sqlalchemy.dialects.postgresql import TSVECTOR from sqlalchemy import select, cast select(cast("some text", TSVECTOR))
产生一个等效于的语句:
SELECT CAST('some text' AS TSVECTOR) AS anon_1
func
命名空间由 PostgreSQL 方言增强,以设置大多数全文搜索函数的正确参数和返回类型。假定已导入 sqlalchemy.dialects.postgresql
包,或者使用 postgresql
方言调用了 create_engine()
,这些函数将自动由 sqlalchemy.sql.expression.func
命名空间使用。这些函数在以下文档中有详细说明:
to_tsvector
to_tsquery
plainto_tsquery
phraseto_tsquery
websearch_to_tsquery
ts_headline
使用 match()
或自定义操作符指定“regconfig”
PostgreSQL 的 plainto_tsquery()
函数接受一个可选的“regconfig”参数,用于指示 PostgreSQL 使用特定的预计��� GIN 或 GiST 索引来执行搜索。在使用 Operators.match()
时,可以使用 postgresql_regconfig
参数指定此附加参数,例如:
select(mytable.c.id).where( mytable.c.title.match('somestring', postgresql_regconfig='english') )
会发出:
SELECT mytable.id FROM mytable WHERE mytable.title @@ plainto_tsquery('english', 'somestring')
使用其他 PostgreSQL 搜索函数与 func
一起使用时,“regconfig” 参数可以直接作为初始参数传递:
select(mytable.c.id).where( func.to_tsvector("english", mytable.c.title).bool_op("@@")( func.to_tsquery("english", "somestring") ) )
生成等效语句:
SELECT mytable.id FROM mytable WHERE to_tsvector('english', mytable.title) @@ to_tsquery('english', 'somestring')
建议使用 PostgreSQL 的 EXPLAIN ANALYZE...
工具,以确保您正在生成利用为全文搜索创建的任何索引的 SQLAlchemy 查询。
另请参阅
全文搜索 - PostgreSQL 文档中的内容
仅来自…
该方言支持 PostgreSQL 的 ONLY 关键字,用于仅针对继承层次结构中的特定表。这可用于生成 SELECT ... FROM ONLY
、UPDATE ONLY ...
和 DELETE FROM ONLY ...
语法。它使用 SQLAlchemy 的提示机制:
# SELECT ... FROM ONLY ... result = table.select().with_hint(table, 'ONLY', 'postgresql') print(result.fetchall()) # UPDATE ONLY ... table.update(values=dict(foo='bar')).with_hint('ONLY', dialect_name='postgresql') # DELETE FROM ONLY ... table.delete().with_hint('ONLY', dialect_name='postgresql')
PostgreSQL 特定的索引选项
有几个针对 PostgreSQL 方言的 Index
构造的扩展可用。
覆盖索引
postgresql_include
选项为给定的字符串名称呈现 INCLUDE(colname):
Index("my_index", table.c.x, postgresql_include=['y'])
将索引呈现为 CREATE INDEX my_index ON table (x) INCLUDE (y)
请注意,此功能需要 PostgreSQL 11 或更高版本。
版本 1.4 中的新功能。
部分索引
部分索引向索引定义添加条件,以便将索引应用于行的子集。这些可以在 Index
上使用 postgresql_where
关键字参数指定:
Index('my_index', my_table.c.id, postgresql_where=my_table.c.value > 10) ```### 操作符类 PostgreSQL 允许为索引的每一列指定一个 *操作符类*(参见 [`www.postgresql.org/docs/current/interactive/indexes-opclass.html`](https://www.postgresql.org/docs/current/interactive/indexes-opclass.html))。`Index` 构造允许通过 `postgresql_ops` 关键字参数指定这些内容: ```py Index( 'my_index', my_table.c.id, my_table.c.data, postgresql_ops={ 'data': 'text_pattern_ops', 'id': 'int4_ops' })
请注意,postgresql_ops
字典中的键是 Column
的“键”名称,即从 .c
集合中访问它的名称,它可以配置为与数据库中实际列的名称不同。
如果要对复杂的 SQL 表达式(例如函数调用)使用 postgresql_ops
,那么它必须被赋予一个在字典中由名称标识的标签,例如:
Index( 'my_index', my_table.c.id, func.lower(my_table.c.data).label('data_lower'), postgresql_ops={ 'data_lower': 'text_pattern_ops', 'id': 'int4_ops' })
操作符类也支持 ExcludeConstraint
结构,使用 ExcludeConstraint.ops
参数。有关详细信息,请参阅该参数。
新版本 1.3.21 中新增对 ExcludeConstraint
的操作符类的支持。
索引类型
PostgreSQL 提供了几种索引类型:B-Tree、Hash、GiST 和 GIN,以及用户创建自己的能力(参见www.postgresql.org/docs/current/static/indexes-types.html
)。这些可以在 Index
上使用 postgresql_using
关键字参数指定:
Index('my_index', my_table.c.data, postgresql_using='gin')
传递给关键字参数的值将简单地传递到底层 CREATE INDEX 命令,因此它 必须 是你的 PostgreSQL 版本的有效索引类型。
索引存储参数
PostgreSQL 允许在索引上设置存储参数。可用的存储参数取决于索引使用的索引方法。存储参数可以在 Index
上使用 postgresql_with
关键字参数指定:
Index('my_index', my_table.c.data, postgresql_with={"fillfactor": 50})
PostgreSQL 允许在创建索引的表空间中定义表空间。表空间可以在 Index
上使用 postgresql_tablespace
关键字参数指定:
Index('my_index', my_table.c.data, postgresql_tablespace='my_tablespace')
注意,同样的选项也可在 Table
上使用。### 使用 CONCURRENTLY 的索引
支持 PostgreSQL 索引选项 CONCURRENTLY,通过将标志 postgresql_concurrently
传递给 Index
结构:
tbl = Table('testtbl', m, Column('data', Integer)) idx1 = Index('test_idx1', tbl.c.data, postgresql_concurrently=True)
上述索引结构将呈现 DDL 用于 CREATE INDEX,假设检测到 PostgreSQL 8.2 或更高版本,或者对于无连接的方言,如下所示:
CREATE INDEX CONCURRENTLY test_idx1 ON testtbl (data)
对于 DROP INDEX,假设检测到 PostgreSQL 9.2 或更高版本,或者对于无连接的方言,它将发出:
DROP INDEX CONCURRENTLY test_idx1
在使用 CONCURRENTLY 时,PostgreSQL 数据库要求该语句在事务块外调用。即使对于单个语句,Python DBAPI 也要求存在事务,因此要使用此结构,必须使用 DBAPI 的“autocommit”模式:
metadata = MetaData() table = Table( "foo", metadata, Column("id", String)) index = Index( "foo_idx", table.c.id, postgresql_concurrently=True) with engine.connect() as conn: with conn.execution_options(isolation_level='AUTOCOMMIT'): table.create(conn)
另请参见
事务隔离级别
覆盖索引
postgresql_include
选项为给定的字符串名称呈现 INCLUDE(colname):
Index("my_index", table.c.x, postgresql_include=['y'])
将索引呈现为CREATE INDEX my_index ON table (x) INCLUDE (y)
注意,此功能要求 PostgreSQL 11 或更高版本。
版本 1.4 中的新功能。
部分索引
部分索引向索引定义添加条件,以便将索引应用于行的子集。这些可以在Index
上使用postgresql_where
关键字参数指定:
Index('my_index', my_table.c.id, postgresql_where=my_table.c.value > 10)
操作类
PostgreSQL 允许为索引的每列指定操作类(参见www.postgresql.org/docs/current/interactive/indexes-opclass.html
)。Index
构造允许通过postgresql_ops
���键字参数指定这些:
Index( 'my_index', my_table.c.id, my_table.c.data, postgresql_ops={ 'data': 'text_pattern_ops', 'id': 'int4_ops' })
注意postgresql_ops
字典中的键是Column
的“键”名称,即从Table
的.c
集合中访问它所使用的名称,这个名称可以配置为与数据库中实际列名不同。
如果要针对复杂的 SQL 表达式(如函数调用)使用postgresql_ops
,则必须为列指定一个在字典中以名称标识的标签,例如:
Index( 'my_index', my_table.c.id, func.lower(my_table.c.data).label('data_lower'), postgresql_ops={ 'data_lower': 'text_pattern_ops', 'id': 'int4_ops' })
操作类也受ExcludeConstraint
构造的支持,使用ExcludeConstraint.ops
参数。查看该参数以获取详细信息。
版本 1.3.21 中的新功能:增加了对ExcludeConstraint
的操作类支持。
索引类型
PostgreSQL 提供了几种索引类型:B-Tree、Hash、GiST 和 GIN,用户还可以创建自己的索引类型(参见www.postgresql.org/docs/current/static/indexes-types.html
)。这些可以在Index
上使用postgresql_using
关键字参数指定:
Index('my_index', my_table.c.data, postgresql_using='gin')
传递给关键字参数的值将简单地传递到底层的 CREATE INDEX 命令,因此它必须是您的 PostgreSQL 版本的有效索引类型。
索引存储参数
PostgreSQL 允许在索引上设置存储参数。可用的存储参数取决于索引使用的索引方法。可以使用 postgresql_with
关键字参数在 Index
上指定存储参数:
Index('my_index', my_table.c.data, postgresql_with={"fillfactor": 50})
PostgreSQL 允许在其中创建索引的表空间中定义表空间。可以使用 postgresql_tablespace
关键字参数在 Index
上指定表空间:
Index('my_index', my_table.c.data, postgresql_tablespace='my_tablespace')
注意,相同的选项也适用于 Table
。
CONCURRENTLY 索引
通过将标志 postgresql_concurrently
传递给 Index
构造来支持 PostgreSQL 的索引选项 CONCURRENTLY:
tbl = Table('testtbl', m, Column('data', Integer)) idx1 = Index('test_idx1', tbl.c.data, postgresql_concurrently=True)
上述索引构造将生成 DDL 用于 CREATE INDEX,假设检测到 PostgreSQL 8.2 或更高版本,或者对于无连接的方言,如下所示:
CREATE INDEX CONCURRENTLY test_idx1 ON testtbl (data)
对于 DROP INDEX,假设检测到 PostgreSQL 9.2 或更高版本,或者对于无连接的方言,它将生成:
DROP INDEX CONCURRENTLY test_idx1
在使用 CONCURRENTLY 时,PostgreSQL 数据库要求该语句在事务块外部调用。Python DBAPI 强制即使对于单个语句,也必须存在事务,因此要使用此构造,必须使用 DBAPI 的“自动提交”模式:
metadata = MetaData() table = Table( "foo", metadata, Column("id", String)) index = Index( "foo_idx", table.c.id, postgresql_concurrently=True) with engine.connect() as conn: with conn.execution_options(isolation_level='AUTOCOMMIT'): table.create(conn)
参见
事务隔离级别
PostgreSQL 索引反射
当使用唯一约束构造时,PostgreSQL 数据库会隐式地创建一个唯一索引。当使用Inspector
检查表时,Inspector.get_indexes()
和Inspector.get_unique_constraints()
会分别报告这两个构造;在索引的情况下,如果检测到与约束相对应,则索引条目中将存在键duplicates_constraint
。当使用Table(..., autoload_with=engine)
进行反射时,如果检测到索引与Table.constraints
集合中的UniqueConstraint
相对应,则唯一索引不会在Table.indexes
中返回。
特殊反射选���
用于 PostgreSQL 后端的Inspector
是PGInspector
的一个实例,提供了额外的方法:
from sqlalchemy import create_engine, inspect engine = create_engine("postgresql+psycopg2://localhost/test") insp = inspect(engine) # will be a PGInspector print(insp.get_enums())
class sqlalchemy.dialects.postgresql.base.PGInspector
成员
get_domains(), get_enums(), get_foreign_table_names(), get_table_oid(), has_type()
类签名
类sqlalchemy.dialects.postgresql.base.PGInspector
(sqlalchemy.engine.reflection.Inspector
)
method get_domains(schema: str | None = None) → List[ReflectedDomain]
返回一个域对象的列表。
每个成员都是一个包含以下字段的字典:
- name - 域的名称
- schema - 域的模式名称。
- visible - 布尔值,表示此域是否在默认搜索路径中可见。
- type - 此域定义的类型。
- nullable - 指示此域是否可以为
NULL
。- default - 域的默认值,如果域没有默认值,则为
None
。- 约束 - 由此域定义的约束的字典列表。每个元素包含两个键:约束的
name
和check
约束文本。
参数:
schema – 模式名称。如果为 None,则使用默认模式(通常为’public’)。也可以设置为'*'
以表示加载所有模式的域。
2.0 版本中的新内容。
method get_enums(schema: str | None = None) → List[ReflectedEnum]
返回 ENUM 对象的列表。
每个成员都是一个包含以下字段的字典:
- name - 枚举的名称
- schema - 枚举的模式名称。
- visible - 布尔值,此枚举是否在默认搜索路径中可见。
- labels - 应用于枚举的字符串标签列表。
参数:
schema – 模式名称。如果为 None,则使用默认模式(通常为’public’)。也可以设置为'*'
以表示加载所有模式的枚举。
method get_foreign_table_names(schema: str | None = None) → List[str]
返回 FOREIGN TABLE 名称列表。
其行为类似于Inspector.get_table_names()
,只不过列表仅限于那些报告relkind
值为f
的表。
method get_table_oid(table_name: str, schema: str | None = None) → int
返回给定表名的 OID。
参数:
table_name
– 表的字符串名称。对于特殊引用,请使用quoted_name
。schema
– 字符串模式名称;如果省略,使用数据库连接的默认模式。对于特殊引用,请使用quoted_name
。
method has_type(type_name: str, schema: str | None = None, **kw: Any) → bool
返回数据库是否在提供的模式中具有指定类型。
参数:
type_name
– 要检查的类型。schema
– 模式名称。如果为 None,则使用默认模式(通常为’public’)。也可以设置为'*'
以在所有模式中检查。
2.0 版本中的新内容。
PostgreSQL 表选项
PostgreSQL 方言与Table
结构直接支持 CREATE TABLE 的几个选项:
INHERITS
:
Table("some_table", metadata, ..., postgresql_inherits="some_supertable") Table("some_table", metadata, ..., postgresql_inherits=("t1", "t2", ...))
ON COMMIT
:
Table("some_table", metadata, ..., postgresql_on_commit='PRESERVE ROWS')
PARTITION BY
:
Table("some_table", metadata, ..., postgresql_partition_by='LIST (part_column)') .. versionadded:: 1.2.6
TABLESPACE
:
Table("some_table", metadata, ..., postgresql_tablespace='some_tablespace')
- 上述选项也适用于
Index
结构。 USING
:
Table("some_table", metadata, ..., postgresql_using='heap') .. versionadded:: 2.0.26
WITH OIDS
:
Table("some_table", metadata, ..., postgresql_with_oids=True)
WITHOUT OIDS
:
Table("some_table", metadata, ..., postgresql_with_oids=False)
参见
PostgreSQL CREATE TABLE options - PostgreSQL 文档中的内容。
PostgreSQL 约束选项
PostgreSQL 方言与选定的约束结构一起支持以下选项:
NOT VALID
:此选项适用于向现有表通过 ALTER TABLE 添加约束时,约束正在添加到约束操作期间不扫描现有行的 CHECK 和 FOREIGN KEY 约束,并且具有的效果。
当使用 SQL 迁移工具(如Alembic)渲染 ALTER TABLE 构造时,postgresql_not_valid
参数可以作为创建约束的操作中的额外关键字参数指定,如下面的 Alembic 示例:
def update(): op.create_foreign_key( "fk_user_address", "address", "user", ["user_id"], ["id"], postgresql_not_valid=True )
- 该关键字最终直接被
CheckConstraint
、ForeignKeyConstraint
和ForeignKey
构造接受;当使用像 Alembic 这样的工具时,特定于方言的关键字参数从迁移操作指令传递给这些构造:
CheckConstraint("some_field IS NOT NULL", postgresql_not_valid=True) ForeignKeyConstraint(["some_id"], ["some_table.some_id"], postgresql_not_valid=True)
- 版本 1.4.32 中的新功能。
另请参见
PostgreSQL ALTER TABLE 选项 - PostgreSQL 文档中的内容。
表值、表和列值函数、行和元组对象
PostgreSQL 在现代 SQL 形式中大量使用表值函数、表和行作为值。这些构造通常作为 PostgreSQL 对复杂数据类型(如 JSON、ARRAY 和其他数据类型)支持的一部分。SQLAlchemy 的 SQL 表达式语言对大多数表值和行值形式具有原生支持。
表值函数
许多 PostgreSQL 内置函数旨在在 SELECT 语句的 FROM 子句中使用,并能够返回表行或表行集。例如,PostgreSQL 的许多 JSON 函数,如 json_array_elements()
、json_object_keys()
、json_each_text()
、json_each()
、json_to_record()
、json_populate_recordset()
使用这些形式。在 SQLAlchemy 中,通过与从func
命名空间生成的Function
对象一起使用FunctionElement.table_valued()
方法,可以使用这些类别的 SQL 函数调用形式。
下面是来自 PostgreSQL 参考文档的示例:
json_each()
:
>>> from sqlalchemy import select, func >>> stmt = select(func.json_each('{"a":"foo", "b":"bar"}').table_valued("key", "value")) >>> print(stmt) SELECT anon_1.key, anon_1.value FROM json_each(:json_each_1) AS anon_1
json_populate_record()
:
>>> from sqlalchemy import select, func, literal_column >>> stmt = select( ... func.json_populate_record( ... literal_column("null::myrowtype"), ... '{"a":1,"b":2}' ... ).table_valued("a", "b", name="x") ... ) >>> print(stmt) SELECT x.a, x.b FROM json_populate_record(null::myrowtype, :json_populate_record_1) AS x
json_to_record()
- 这种形式使用了 PostgreSQL 特定的派生列形式在别名中,我们可以利用column()
元素与类型来生成它们。FunctionElement.table_valued()
方法生成了一个TableValuedAlias
构造,并且方法TableValuedAlias.render_derived()
设置了派生列规范:
>>> from sqlalchemy import select, func, column, Integer, Text >>> stmt = select( ... func.json_to_record('{"a":1,"b":[1,2,3],"c":"bar"}').table_valued( ... column("a", Integer), column("b", Text), column("d", Text), ... ).render_derived(name="x", with_types=True) ... ) >>> print(stmt) SELECT x.a, x.b, x.d FROM json_to_record(:json_to_record_1) AS x(a INTEGER, b TEXT, d TEXT)
WITH ORDINALITY
- SQL 标准的一部分,WITH ORDINALITY
将一个序号计数器添加到函数的输出中,并被一组有限的 PostgreSQL 函数接受,包括unnest()
和generate_series()
。FunctionElement.table_valued()
方法接受一个名为with_ordinality
的关键字参数,用于此目的,该参数接受将应用于“序号”列的字符串名称:
>>> from sqlalchemy import select, func >>> stmt = select( ... func.generate_series(4, 1, -1). ... table_valued("value", with_ordinality="ordinality"). ... render_derived() ... ) >>> print(stmt) SELECT anon_1.value, anon_1.ordinality FROM generate_series(:generate_series_1, :generate_series_2, :generate_series_3) WITH ORDINALITY AS anon_1(value, ordinality)
1.4.0b2 版本中的新功能。
另请参阅
表值函数 - 在 SQLAlchemy 统一教程 中 ### 列值函数
类似于表值函数,列值函数存在于 FROM 子句中,但将自身提供给列子句作为单个标量值。PostgreSQL 函数,如 json_array_elements()
、unnest()
和 generate_series()
可以使用这种形式。使用 FunctionElement.column_valued()
方法的 FunctionElement
可用于列值函数:
json_array_elements()
:
>>> from sqlalchemy import select, func >>> stmt = select(func.json_array_elements('["one", "two"]').column_valued("x")) >>> print(stmt) SELECT x FROM json_array_elements(:json_array_elements_1) AS x
unnest()
- 为了生成 PostgreSQL ARRAY 文字,可以使用array()
构造:
>>> from sqlalchemy.dialects.postgresql import array >>> from sqlalchemy import select, func >>> stmt = select(func.unnest(array([1, 2])).column_valued()) >>> print(stmt) SELECT anon_1 FROM unnest(ARRAY[%(param_1)s, %(param_2)s]) AS anon_1
- 当然,该函数可以用于已存在的类型为
ARRAY
的表列:
>>> from sqlalchemy import table, column, ARRAY, Integer >>> from sqlalchemy import select, func >>> t = table("t", column('value', ARRAY(Integer))) >>> stmt = select(func.unnest(t.c.value).column_valued("unnested_value")) >>> print(stmt) SELECT unnested_value FROM unnest(t.value) AS unnested_value
另请参阅
列值函数 - 表值函数作为标量列 - 在 SQLAlchemy 统一教程
行类型
使用sqlalchemy.func
命名空间中的func.ROW
或者使用tuple_()
构造函数,可以近似实现对ROW
的渲染支持:
>>> from sqlalchemy import table, column, func, tuple_ >>> t = table("t", column("id"), column("fk")) >>> stmt = t.select().where( ... tuple_(t.c.id, t.c.fk) > (1,2) ... ).where( ... func.ROW(t.c.id, t.c.fk) < func.ROW(3, 7) ... ) >>> print(stmt) SELECT t.id, t.fk FROM t WHERE (t.id, t.fk) > (:param_1, :param_2) AND ROW(t.id, t.fk) < ROW(:ROW_1, :ROW_2)
另请参阅
传递给函数的表类型
PostgreSQL 支持将表作为函数的参数传递,这被称为“record”类型。SQLAlchemy 的FromClause
对象,如Table
,支持使用FromClause.table_valued()
方法的这种特殊形式,该方法类似于FunctionElement.table_valued()
方法,只是列集合已经由FromClause
本身建立:
>>> from sqlalchemy import table, column, func, select >>> a = table( "a", column("id"), column("x"), column("y")) >>> stmt = select(func.row_to_json(a.table_valued())) >>> print(stmt) SELECT row_to_json(a) AS row_to_json_1 FROM a
在版本 1.4.0b2 中新增。
表值函数
许多 PostgreSQL 内置函数旨在在 SELECT 语句的 FROM 子句中使用,并且能够返回表行或表行集。例如,PostgreSQL 的许多 JSON 函数,如json_array_elements()
、json_object_keys()
、json_each_text()
、json_each()
、json_to_record()
、json_populate_recordset()
等,使用了这种形式。在 SQLAlchemy 中,这些类 SQL 函数调用形式可通过FunctionElement.table_valued()
方法与从func
命名空间生成的Function
对象一起使用。
下面是来自 PostgreSQL 参考文档的示例:
json_each()
:
>>> from sqlalchemy import select, func >>> stmt = select(func.json_each('{"a":"foo", "b":"bar"}').table_valued("key", "value")) >>> print(stmt) SELECT anon_1.key, anon_1.value FROM json_each(:json_each_1) AS anon_1
json_populate_record()
:
>>> from sqlalchemy import select, func, literal_column >>> stmt = select( ... func.json_populate_record( ... literal_column("null::myrowtype"), ... '{"a":1,"b":2}' ... ).table_valued("a", "b", name="x") ... ) >>> print(stmt) SELECT x.a, x.b FROM json_populate_record(null::myrowtype, :json_populate_record_1) AS x
json_to_record()
- 此形式使用了 PostgreSQL 特定的派生列形式在别名中,我们可以利用column()
元素和类型来产生它们。FunctionElement.table_valued()
方法产生了一个TableValuedAlias
结构,而方法TableValuedAlias.render_derived()
设置了派生列的规范:
>>> from sqlalchemy import select, func, column, Integer, Text >>> stmt = select( ... func.json_to_record('{"a":1,"b":[1,2,3],"c":"bar"}').table_valued( ... column("a", Integer), column("b", Text), column("d", Text), ... ).render_derived(name="x", with_types=True) ... ) >>> print(stmt) SELECT x.a, x.b, x.d FROM json_to_record(:json_to_record_1) AS x(a INTEGER, b TEXT, d TEXT)
WITH ORDINALITY
- SQL 标准的一部分,WITH ORDINALITY
向函数的输出添加一个序数计数器,并被一组有限的 PostgreSQL 函数接受,包括unnest()
和generate_series()
。FunctionElement.table_valued()
方法接受一个名为with_ordinality
的关键字参数,用于此目的,该参数接受将应用于“序数”列的字符串名称:
>>> from sqlalchemy import select, func >>> stmt = select( ... func.generate_series(4, 1, -1). ... table_valued("value", with_ordinality="ordinality"). ... render_derived() ... ) >>> print(stmt) SELECT anon_1.value, anon_1.ordinality FROM generate_series(:generate_series_1, :generate_series_2, :generate_series_3) WITH ORDINALITY AS anon_1(value, ordinality)
1.4.0b2 版本中新增。
另请参阅
表值函数 - 在 SQLAlchemy 统一教程 中
列值函数
类似于表值函数,列值函数出现在 FROM 子句中,但将其自身交付给列子句作为单个标量值。PostgreSQL 函数如 json_array_elements()
、unnest()
和 generate_series()
可以使用此形式。列值函数可通过 FunctionElement.column_valued()
方法的 FunctionElement
来使用:
json_array_elements()
:
>>> from sqlalchemy import select, func >>> stmt = select(func.json_array_elements('["one", "two"]').column_valued("x")) >>> print(stmt) SELECT x FROM json_array_elements(:json_array_elements_1) AS x
unnest()
- 为了生成 PostgreSQL 的数组文字,可以使用array()
结构:
>>> from sqlalchemy.dialects.postgresql import array >>> from sqlalchemy import select, func >>> stmt = select(func.unnest(array([1, 2])).column_valued()) >>> print(stmt) SELECT anon_1 FROM unnest(ARRAY[%(param_1)s, %(param_2)s]) AS anon_1
- 当然,该函数也可以用于已存在的类型为
ARRAY
的表绑定列:
>>> from sqlalchemy import table, column, ARRAY, Integer >>> from sqlalchemy import select, func >>> t = table("t", column('value', ARRAY(Integer))) >>> stmt = select(func.unnest(t.c.value).column_valued("unnested_value")) >>> print(stmt) SELECT unnested_value FROM unnest(t.value) AS unnested_value
另请参阅
列值函数 - 作为标量列的表值函数 - 在 SQLAlchemy 统一教程 中
行类型
内置支持渲染 ROW
可以使用 func.ROW
和 sqlalchemy.func
命名空间来近似,或者使用 tuple_()
构造:
>>> from sqlalchemy import table, column, func, tuple_ >>> t = table("t", column("id"), column("fk")) >>> stmt = t.select().where( ... tuple_(t.c.id, t.c.fk) > (1,2) ... ).where( ... func.ROW(t.c.id, t.c.fk) < func.ROW(3, 7) ... ) >>> print(stmt) SELECT t.id, t.fk FROM t WHERE (t.id, t.fk) > (:param_1, :param_2) AND ROW(t.id, t.fk) < ROW(:ROW_1, :ROW_2)
另请参阅
传递给函数的表类型
PostgreSQL 支持将表作为函数的参数传递,这称为“record”类型。SQLAlchemy FromClause
对象,例如 Table
支持使用 FromClause.table_valued()
方法来实现此特殊形式,该方法类似于 FunctionElement.table_valued()
方法,只是列的集合已由 FromClause
本身建立:
>>> from sqlalchemy import table, column, func, select >>> a = table( "a", column("id"), column("x"), column("y")) >>> stmt = select(func.row_to_json(a.table_valued())) >>> print(stmt) SELECT row_to_json(a) AS row_to_json_1 FROM a
1.4.0b2 版本中的新功能。
数组类型
PostgreSQL 方言支持数组,既作为多维列类型,也作为数组文本:
ARRAY
- 数组数据类型array
- 数组文本array_agg()
- ARRAY_AGG SQL 函数aggregate_order_by
- PG 的 ORDER BY 聚合函数语法的辅助工具。
JSON 类型
PostgreSQL 方言支持 JSON 和 JSONB 数据类型,包括 psycopg2 的本机支持以及对 PostgreSQL 的所有特殊操作符的支持:
JSON
JSONB
JSONPATH
HSTORE 类型
PostgreSQL 的 HSTORE 类型以及 hstore 文本都受支持:
HSTORE
- HSTORE 数据类型hstore
- hstore 文本
枚举类型
PostgreSQL 具有独立可创建的 TYPE 结构,用于实现枚举类型。这种方法在 SQLAlchemy 方面引入了显着的复杂性,涉及何时应该创建和删除此类型。类型对象也是一个独立的可反射实体。应查阅以下部分:
ENUM
- 用于 ENUM 的 DDL 和类型支持。PGInspector.get_enums()
- 检索当前 ENUM 类型的列表ENUM.create()
,ENUM.drop()
- 用于 ENUM 的单独 CREATE 和 DROP 命令。
使用 ENUM 与 ARRAY
目前后端 DBAPI 不直接支持 ENUM 和 ARRAY 的组合。在 SQLAlchemy 1.3.17 之前,需要特殊的变通方法才能使此组合工作,如下所述。
自版本 1.3.17 起更改:现在 SQLAlchemy 的实现直接处理 ENUM 和 ARRAY 的组合,无需任何需要的变通方法。
from sqlalchemy import TypeDecorator from sqlalchemy.dialects.postgresql import ARRAY class ArrayOfEnum(TypeDecorator): impl = ARRAY def bind_expression(self, bindvalue): return sa.cast(bindvalue, self) def result_processor(self, dialect, coltype): super_rp = super(ArrayOfEnum, self).result_processor(dialect, coltype) def handle_raw_string(value): inner = re.match(r"^{(.*)}$", value).group(1) return inner.split(",") if inner else [] def process(value): if value is None: return None return super_rp(handle_raw_string(value)) return process
例如:
Table( "mydata", metadata, Column("id", Integer, primary_key=True), Column("data", ArrayOfEnum(ENUM("a", "b", "c", name="myenum"))), )
此类型未作为内置类型包含,因为它与突然决定在新版本中直接支持 ENUM 的 DBAPI 不兼容。 ### 使用 JSON/JSONB 与 ARRAY
类似于使用 ENUM,在 SQLAlchemy 1.3.17 之前,对于 JSON/JSONB 的 ARRAY,我们需要呈现适当的 CAST。当前的 psycopg2 驱动程序可以正确地处理结果集,无需任何特殊步骤。
自版本 1.3.17 起更改:现在 SQLAlchemy 的实现直接处理 JSON/JSONB 和 ARRAY 的组合,无需任何需要的变通方法。
class CastingArray(ARRAY): def bind_expression(self, bindvalue): return sa.cast(bindvalue, self)
例如:
Table( "mydata", metadata, Column("id", Integer, primary_key=True), Column("data", CastingArray(JSONB)), ) ```### 使用 ENUM 与 ARRAY 目前后端 DBAPI 不直接支持 ENUM 和 ARRAY 的组合。在 SQLAlchemy 1.3.17 之前,需要特殊的变通方法才能使此组合工作,如下所述。 自版本 1.3.17 起更改:现在 SQLAlchemy 的实现直接处理 ENUM 和 ARRAY 的组合,无需任何需要的变通方法。 ```py from sqlalchemy import TypeDecorator from sqlalchemy.dialects.postgresql import ARRAY class ArrayOfEnum(TypeDecorator): impl = ARRAY def bind_expression(self, bindvalue): return sa.cast(bindvalue, self) def result_processor(self, dialect, coltype): super_rp = super(ArrayOfEnum, self).result_processor(dialect, coltype) def handle_raw_string(value): inner = re.match(r"^{(.*)}$", value).group(1) return inner.split(",") if inner else [] def process(value): if value is None: return None return super_rp(handle_raw_string(value)) return process
例如:
Table( "mydata", metadata, Column("id", Integer, primary_key=True), Column("data", ArrayOfEnum(ENUM("a", "b", "c", name="myenum"))), )
此类型未作为内置类型包含,因为它与突然决定在新版本中直接支持 ENUM 的 DBAPI 不兼容。
使用 JSON/JSONB 与 ARRAY
类似于使用 ENUM,在 SQLAlchemy 1.3.17 之前,对于 JSON/JSONB 的 ARRAY,我们需要呈现适当的 CAST。当前的 psycopg2 驱动程序可以正确地处理结果集,无需任何特殊步骤。
自版本 1.3.17 起更改:现在 SQLAlchemy 的实现直接处理 JSON/JSONB 和 ARRAY 的组合,无需任何需要的变通方法。
class CastingArray(ARRAY): def bind_expression(self, bindvalue): return sa.cast(bindvalue, self)
例如:
Table( "mydata", metadata, Column("id", Integer, primary_key=True), Column("data", CastingArray(JSONB)), )
SqlAlchemy 2.0 中文文档(四十八)(4)https://developer.aliyun.com/article/1563010