SqlAlchemy 2.0 中文文档（四十八）（3）-阿里云开发者社区

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 行构造函数

PostgreSQL 行构造函数比较

传递给函数的表类型

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 行构造函数

PostgreSQL 行构造函数比较

传递给函数的表类型

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

自版本 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

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

SqlAlchemy 2.0 中文文档（四十八）（3）

事务隔离级别

设置只读 / 可延迟

用于连接池的临时表 / 资源重置

在连接时设置备用搜索路径

远程模式表内省和 PostgreSQL search_path

INSERT/UPDATE…RETURNING

INSERT…ON CONFLICT（插入或更新）

指定目标

SET 子句

使用被排除的插入值进行更新

附加的 WHERE 条件

使用 DO NOTHING 跳过行

指定目标

SET 子句

使用排除的插入值进行更新

附加的 WHERE 条件

使用 DO NOTHING 跳过行

全文搜索

使用 match() 进行简单纯文本���配

直接使用 PostgreSQL 全文函数和运算符

使用match()或自定义运算符指定“regconfig”

使用 match() 进行简单纯文本匹配

直接使用 PostgreSQL 全文搜索函数和运算符

使用 match() 或自定义操作符指定“regconfig”

仅来自…

PostgreSQL 特定的索引选项

覆盖索引

部分索引

索引类型

索引存储参数

覆盖索引

部分索引

操作类

索引类型

索引存储参数

CONCURRENTLY 索引

PostgreSQL 索引反射

特殊反射选���

PostgreSQL 表选项

PostgreSQL 约束选项

表值、表和列值函数、行和元组对象

表值函数

行类型

传递给函数的表类型

表值函数

列值函数

行类型

传递给函数的表类型

数组类型

JSON 类型

HSTORE 类型

枚举类型

使用 ENUM 与 ARRAY

使用 JSON/JSONB 与 ARRAY

热门文章

最新文章

相关电子书

使用 `match()` 进行简单纯文本��配

使用`match()`或自定义运算符指定“regconfig”

使用 `match()` 进行简单纯文本匹配

使用 `match()` 或自定义操作符指定“regconfig”

特殊反射选��