方言
方言 是 SQLAlchemy 用来与各种类型的 DBAPI 实现和数据库通信的系统。以下各节包含每个后端使用的特定用法的参考文档和说明,以及各种 DBAPI 的说明。
所有方言都要求安装适当的 DBAPI 驱动程序。
包含的方言
- PostgreSQL
- MySQL 和 MariaDB
- SQLite
- Oracle
- Microsoft SQL Server
包含的方言的支持级别
以下表格总结了每个包含的方言的支持级别。
包含的方言支持的数据库版本
数据库 | 在 CI 中完全测试 | 正常支持 | 尽力而为 |
Microsoft SQL Server | 2017 | 2012+ | 2005+ |
MySQL / MariaDB | 5.6, 5.7, 8.0 / 10.8, 10.9 | 5.6+ / 10+ | 5.0.2+ / 5.0.2+ |
Oracle | 18c | 11+ | 9+ |
PostgreSQL | 12, 13, 14, 15 | 9.6+ | 9+ |
SQLite | 3.36.0 | 3.12+ | 3.7.16+ |
支持定义
在 CI 中完全测试
在 CI 中完全测试 表示已在 sqlalchemy CI 系统中进行测试并通过测试套件中的所有测试的版本。
正常支持
正常支持 表示大多数功能应该可以工作,但不是所有版本都在 ci 配置中进行测试,因此可能存在一些不受支持的边缘情况。我们将尝试修复影响这些版本的问题。
尽力而为
尽力而为 指我们尝试在其中支持基本功能,但在某些用例中可能存在不受支持的功能或错误。我们可能会接受相关问题的拉取请求以继续支持较旧的版本,这些请求会逐案审查。 ## 外部方言
目前为 SQLAlchemy 维护的外部方言项目包括:
| YugabyteDB | sqlalchemy-yugabytedb | ## 包含的方言
- PostgreSQL
- MySQL 和 MariaDB
- SQLite
- Oracle
- Microsoft SQL Server
包含的方言的支持级别
下表总结了每个包含的方言的支持级别。
包含的方言的支持数据库版本
数据库 | 在 CI 中完全测试过 | 普通支持 | 尽力而为 |
Microsoft SQL Server | 2017 | 2012+ | 2005+ |
MySQL / MariaDB | 5.6、5.7、8.0 / 10.8、10.9 | 5.6+ / 10+ | 5.0.2+ / 5.0.2+ |
Oracle | 18c | 11+ | 9+ |
PostgreSQL | 12、13、14、15 | 9.6+ | 9+ |
SQLite | 3.36.0 | 3.12+ | 3.7.16+ |
支持定义
在 CI 中完全测试过
在 CI 中完全测试过 表示经过在 sqlalchemy CI 系统中测试的版本,并通过测试套件中的所有测试。
普通支持
普通支持 表示大多数功能应该正常工作,但并非所有版本都在 ci 配置中进行测试,因此可能存在一些不受支持的边缘情况。我们将尝试修复影响这些版本的问题。
尽力而为
尽力而为 表示我们尽力在这些版本上支持基本功能,但在某些用例中可能会存在不支持的功能或错误。可能会接受带有相关问题的拉取请求,以继续支持旧版本,这些请求会根据具体情况进行审查。
已包括方言的支持级别
以下表总结了每个已包含方言的支持级别。
已包括方言的支持数据库版本
数据库 | 在 CI 中进行了全面测试 | 普通支持 | 尽力而为 |
Microsoft SQL Server | 2017 | 2012+ | 2005+ |
MySQL / MariaDB | 5.6, 5.7, 8.0 / 10.8, 10.9 | 5.6+ / 10+ | 5.0.2+ / 5.0.2+ |
Oracle | 18c | 11+ | 9+ |
PostgreSQL | 12、13、14、15 | 9.6+ | 9+ |
SQLite | 3.36.0 | 3.12+ | 3.7.16+ |
支持定义
在 CI 中进行了全面测试
在 CI 中进行了全面测试 表示已经在 sqlalchemy CI 系统中测试并通过了测试套件中的所有测试的版本。
普通支持
普通支持 表示大多数功能应该可以正常工作,但并非所有版本都在 ci 配置中进行了测试,因此可能存在一些不受支持的边缘情况。我们将尝试修复影响这些版本的问题。
尽力而为
尽力而为 表明我们尽力在这些版本上支持基本功能,但在某些用例中可能会存在不支持的功能或错误。可能会接受带有相关问题的拉取请求,以继续支持旧版本,这些请求会根据具体情况进行审查。
外部方言
目前由 SQLAlchemy 维护的外部方言项目包括:
PostgreSQL
对 PostgreSQL 数据库的支持。
以下表总结了当前数据库发布版本的支持水平。
支持的 PostgreSQL 版本
支持类型 | 版本 |
CI 中完全测试通过 | 12, 13, 14, 15 |
正常支持 | 9.6+ |
最佳尝试 | 9+ |
DBAPI 支持
以下方言/DBAPI 选项可用。请参考各个 DBAPI 部分获取连接信息。
- psycopg2
- psycopg (又名 psycopg 3)
- pg8000
- asyncpg
- psycopg2cffi
序列/SERIAL/IDENTITY
PostgreSQL 支持序列,而 SQLAlchemy 使用这些作为为整数型主键列创建新主键值的默认方式。在创建表时,SQLAlchemy 将为整数型主键列发出SERIAL
数据类型,生成与列对应的序列和服务器端默认值。
要指定用于主键生成的特定命名序列,请使用Sequence()
构造:
Table( "sometable", metadata, Column( "id", Integer, Sequence("some_id_seq", start=1), primary_key=True ) )
当 SQLAlchemy 发出单个 INSERT 语句时,为了满足“最后插入标识符”可用的约定,将在 INSERT 语句中添加一个 RETURNING 子句,该子句指定在语句完成后应返回主键列。仅当使用 PostgreSQL 8.2 或更高版本时,RETURNING 功能才会生效。作为备用方法,无论是显式指定还是隐式通过SERIAL
指定,序列都会在之前独立执行,返回的值将用于后续插入。请注意,当使用“executemany”语义执行insert()
构造时,“最后插入的标识符”功能不适用;在这种情况下,不会发出 RETURNING 子句,也不会预先执行序列。
PostgreSQL 10 及以上版本的 IDENTITY 列
PostgreSQL 10 及以上版本具有新的 IDENTITY 功能,取代了 SERIAL 的使用。在Column
中的Identity
构造可用于控制其行为:
from sqlalchemy import Table, Column, MetaData, Integer, Computed metadata = MetaData() data = Table( "data", metadata, Column( 'id', Integer, Identity(start=42, cycle=True), primary_key=True ), Column('data', String) )
上述Table
对象的 CREATE TABLE 如下:
CREATE TABLE data ( id INTEGER GENERATED BY DEFAULT AS IDENTITY (START WITH 42 CYCLE), data VARCHAR, PRIMARY KEY (id) )
自版本 1.4 起更改:在Column
中添加Identity
构造,以指定自增列的选项。
注意
以前的 SQLAlchemy 版本不支持渲染 IDENTITY,并可以使用以下编译钩子来将 SERIAL 的出现替换为 IDENTITY:
from sqlalchemy.schema import CreateColumn from sqlalchemy.ext.compiler import compiles @compiles(CreateColumn, 'postgresql') def use_identity(element, compiler, **kw): text = compiler.visit_create_column(element, **kw) text = text.replace( "SERIAL", "INT GENERATED BY DEFAULT AS IDENTITY" ) return text
使用上述方法,一个表如下:
t = Table( 't', m, Column('id', Integer, primary_key=True), Column('data', String) )
将在后端数据库上生成如下:
CREATE TABLE t ( id INT GENERATED BY DEFAULT AS IDENTITY, data VARCHAR, PRIMARY KEY (id) ) ```## 服务器端游标 psycopg2、asyncpg 方言支持服务器端游标支持,其他方言也可能支持。 通过使用`Connection.execution_options.stream_results`连接执行选项,可以在每个语句基础上启用服务器端游标: ```py with engine.connect() as conn: result = conn.execution_options(stream_results=True).execute(text("select * from table"))
请注意,某些类型的 SQL 语句可能不支持服务器端游标;通常,只应使用返回行的 SQL 语句与此选项一起使用。
自版本 1.4 起已弃用:方言级别的server_side_cursors
标志已被弃用,并将在将来的版本中移除。请使用Connection.stream_results
执行选项来支持无缓冲游标。
另请参阅
使用服务器端游标(即流式结果) ## 事务隔离级别
大多数 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 连接对象上的.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 自动提交
设置 READ ONLY / DEFERRABLE
Psycopg2 事务隔离级别
pg8000 事务隔离级别 ## 设置 READ ONLY / DEFERRABLE
大多数 PostgreSQL 方言支持设置事务的 “READ ONLY” 和 “DEFERRABLE” 特性,这是隔离级别设置的补充。通过使用 Connection.execution_options()
方法并传递 postgresql_readonly
和 postgresql_deferrable
标志,可以同时或独立地建立这两个属性与隔离级别。下面的示例说明了在设置 “READ ONLY” 和 “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 隔离级别的 “readonly”。
版本 1.4 中的新功能:增加了对 postgresql_readonly
和 postgresql_deferrable
执行选项的支持。## 临时表 / 资源重置用于连接池
SQLAlchemy Engine
对象使用的 QueuePool
连接池实现包括在连接返回到池时调用 DBAPI 的 .rollback()
方法的 重置行为。虽然此回滚将清除前一个事务使用的即时状态,但它不涵盖更广泛范围的会话级状态,包括临时表以及其他服务器状态,如准备好的语句句柄和语句缓存。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()
事件添加了额外的状态参数,并确保该事件为所有“reset”发生而调用,因此它适合作为自定义“reset”处理程序的地方。以前使用PoolEvents.checkin()
处理程序的方案仍然可用。
另请参见
返回时重置 - 在连接池文档中 ## 在连接上设置备用搜索路径
PostgreSQL 的search_path
变量指的是在 SQL 语句中引用特定表或其他对象时将隐式引用的模式名称列表。如下一节远程模式表内省和 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()
函数指示的模式中,则 Table 的“.schema”属性分配的模式名称是 Python 的“None”值。否则,“.schema”属性将被分配该模式的字符串名称。
关于这些Table
对象通过外键约束引用的表,必须决定在当前search_path
成员也是该远程表的模式名称的情况下,如何在这些远程表中表示.schema
。
默认情况下,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>
上述过程将将referred
表的名称传递给MetaData.tables
集合,不包含模式:
>>> 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 网站上。
插入/更新…返回
该方言支持 PG 8.2 的INSERT..RETURNING
,UPDATE..RETURNING
和DELETE..RETURNING
语法。INSERT..RETURNING
默认用于单行 INSERT 语句,以获取新生成的主键标识符。要指定显式的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())
插入…冲突时执行(Upsert)
从 9.5 版本开始,PostgreSQL 允许通过INSERT
语句的ON CONFLICT
子句将行“upsert”(更新或插入)到表中。只有候选行不违反任何唯一约束时才会插入该行。在唯一约束违反的情况下,可以发生第二个动作,可以是“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
的命名或未命名主键,使用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 类型的 UPDATE,除非它们在 Insert.on_conflict_do_update.set_
字典中手动指定。
使用排除的 INSERT 值进行更新
为了引用建议的插入行,特殊别名 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
而不指定任何列或约束,则会跳过任何唯一或排除约束违规的 INSERT 效果:
>>> 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 方言中,类似以下表达式: ```py 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 搜索函数时,可以直接传递“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
的“键”名称,即从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 允许在索引上设置存储参数。可用的存储参数取决于索引使用的索引方法。存储参数可以在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,通过向 Index
构造传递标志 postgresql_concurrently
:
tbl = Table('testtbl', m, Column('data', Integer)) idx1 = Index('test_idx1', tbl.c.data, postgresql_concurrently=True)
上述索引构造将呈现 DDL 以创建索引,假设检测到 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]
返回 DOMAIN 对象的列表。
每个成员都是包含以下字段的字典:
- 名称 - 领域的名称
- 领域的模式 - 领域的模式名称。
- visible - 布尔值,此领域是否在默认搜索路径中可见。
- 类型 - 此领域定义的类型。
- nullable - 表示此域是否可以为
NULL
。- 默认值 - 域的默认值或
None
(如果域没有默认值)。- 约束 - 包含此域定义的约束的字典列表。每个元素包含两个键:约束的
name
和check
与约束文本。
参数:
模式 – 模式名称。如果为 None,则使用默认模式(通常为 ‘public’)。也可以设置为'*'
以表示加载所有模式的领域。
2.0 版中的新内容。
method get_enums(schema: str | None = None) → List[ReflectedEnum]
返回 ENUM 对象的列表。
每个成员都是包含以下字段的字典:
- 名称 - 枚举的名称
- 枚举的模式 - 枚举的模式名称。
- visible - 布尔值,此枚举是否在默认搜索路径中可见。
- 标签 - 适用于枚举的字符串标签列表。
参数:
模式 – 模式名称。如果为 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 选项 - 在 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 中,使用 FunctionElement.table_valued()
方法与从 func
命名空间生成的 Function
对象,可以使用这些 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()
构造:
>>> 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 统一教程 中
行类型
内置支持以 func.ROW
与 sqlalchemy.func
命名空间来近似渲染 ROW
,或者使用 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 版中新增。
ARRAY 类型
PostgreSQL 方言支持数组,既作为多维列类型,也作为数组字面量:
ARRAY
- 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 字面值
ENUM 类型
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 开始更改:对于 ENUM 和 ARRAY 的组合,现在由 SQLAlchemy 的实现直接处理,无需任何需要的解决方法。
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 的 ARRAY 的 DBAPI 不兼容。### 使用 JSON/JSONB 与 ARRAY
与使用 ENUM 类似,在 SQLAlchemy 1.3.17 之前,对于 JSON/JSONB 数组,我们需要呈现适当的 CAST。当前的 psycopg2 驱动程序可以正确地处理结果集,无需任何特殊步骤。
从版本 1.3.17 开始更改:对于 JSON/JSONB 和 ARRAY 的组合,现在由 SQLAlchemy 的实现直接处理,无需任何需要的解决方法。
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)), ) ```## 范围和多范围类型 PostgreSQL 支持 psycopg、pg8000 和 asyncpg 方言的范围和多范围类型;psycopg2 方言仅支持范围类型。 新版本 2.0.17 中新增了对 pg8000 方言的范围和多范围支持。需要 pg8000 1.29.8 或更高版本。 传递给数据库的数据值可以作为字符串值传递,也可以使用 `Range` 数据对象。 新版本 2.0 中:添加了后端不可知的 `Range` 对象,用于指示范围。`psycopg2` 特定的范围类不再公开,仅在该特定方言内部使用。 例如,使用 `TSRANGE` 数据类型的完全类型化模型示例: ```py from datetime import datetime from sqlalchemy.dialects.postgresql import Range from sqlalchemy.dialects.postgresql import TSRANGE from sqlalchemy.orm import DeclarativeBase from sqlalchemy.orm import Mapped from sqlalchemy.orm import mapped_column class Base(DeclarativeBase): pass class RoomBooking(Base): __tablename__ = "room_booking" id: Mapped[int] = mapped_column(primary_key=True) room: Mapped[str] during: Mapped[Range[datetime]] = mapped_column(TSRANGE)
为了表示上述 during
列的数据,Range
类型是一个简单的数据类,用于表示范围的边界。下面演示了向上述 room_booking
表中插入行的示例:
from sqlalchemy import create_engine from sqlalchemy.orm import Session engine = create_engine("postgresql+psycopg://scott:tiger@pg14/dbname") Base.metadata.create_all(engine) with Session(engine) as session: booking = RoomBooking( room="101", during=Range(datetime(2013, 3, 23), datetime(2013, 3, 25)) ) session.add(booking) session.commit()
从任何范围列中选择也将返回 Range
对象,如下所示:
from sqlalchemy import select with Session(engine) as session: for row in session.execute(select(RoomBooking.during)): print(row)
可用的范围数据类型如下:
INT4RANGE
INT8RANGE
NUMRANGE
DATERANGE
TSRANGE
TSTZRANGE
对象名称 | 描述 |
Range | 表示 PostgreSQL 范围。 |
class sqlalchemy.dialects.postgresql.Range
表示 PostgreSQL 范围。
例如:
r = Range(10, 50, bounds="()")
调用风格类似于 psycopg 和 psycopg2,部分原因是为了更容易从之前直接使用这些对象的 SQLAlchemy 版本迁移。
参数:
lower
– 下限值,或者为 Noneupper
– 上限值,或者为 Nonebounds
– 仅关键字,可选字符串值,为"()"
,"[)"
,"(]"
,"[]"
中的一个。默认为")"
。empty
– 仅关键字,可选布尔值,指示这是一个“空”范围
新版本 2.0 中新增。
成员
[eq(), adjacent_to(), contained_by(), contains(), difference(), intersection(), is_empty, isempty, lower, lower_inc, lower_inf, not_extend_left_of(), not_extend_right_of(), overlaps(), strictly_left_of(), strictly_right_of(), union(), upper, upper_inc, upper_inf
类签名
类sqlalchemy.dialects.postgresql.Range
(typing.Generic
)
method __eq__(other: Any) → bool
考虑边界包含性,将此范围与其他范围进行比较,如果它们相等则返回True
。
method adjacent_to(other: Range[_T]) → bool
确定此范围是否与其他范围相邻。
method contained_by(other: Range[_T]) → bool
确定此范围是否被其他范围包含。
method contains(value: _T | Range[_T]) → bool
确定此范围是否包含值。
method difference(other: Range[_T]) → Range[_T]
计算此范围与其他范围的差异。
如果两个范围“不相交”,即既不相邻也不重叠,则引发ValueError
异常。
method intersection(other: Range[_T]) → Range[_T]
计算此范围与其他范围的交集。
版本 2.0.10 中的新功能。
attribute is_empty
“empty”属性的同义词。
attribute isempty
“empty”属性的同义词。
attribute lower: _T | None
下界
attribute lower_inc
如果下界是包含的,则返回 True。
attribute lower_inf
如果此范围非空且下界为无限,则返回 True。
method not_extend_left_of(other: Range[_T]) → bool
确定此范围是否不向左延伸到其他范围。
method not_extend_right_of(other: Range[_T]) → bool
确定此范围是否不向右延伸到其他范围。
method overlaps(other: Range[_T]) → bool
确定此范围是否与其他范围重叠。
method strictly_left_of(other: Range[_T]) → bool
确定此范围是否完全位于其他范围的左侧。
method strictly_right_of(other: Range[_T]) → bool
确定此范围是否完全位于其他范围的右侧。
method union(other: Range[_T]) → Range[_T]
计算此范围与其他范围的并集。
如果两个范围“不相交”,即既不相邻也不重叠,则引发ValueError
异常。
attribute upper: _T | None
上界
attribute upper_inc
如果上界是包含的,则返回 True。
attribute upper_inf
如果此范围非空且上界为无限,则返回 True。
多范围
PostgreSQL 14 及以上版本支持多范围。SQLAlchemy 的多范围数据类型处理Range
类型的列表。
仅支持在 psycopg、asyncpg 和 pg8000 方言上使用多范围(multiranges)。SQLAlchemy 默认的 postgresql
方言 psycopg2 不支持多范围数据类型。
新版本 2.0 中:增加了对 MULTIRANGE 数据类型的支持。SQLAlchemy 将多范围值表示为 Range
对象的列表。
新版本 2.0.17 中:为 pg8000 方言添加了多范围支持。需要 pg8000 1.29.8 或更高版本。
新版本 2.0.26 中:添加了 MultiRange
序列。
下面的示例说明了使用 TSMULTIRANGE
数据类型:
from datetime import datetime from typing import List from sqlalchemy.dialects.postgresql import Range from sqlalchemy.dialects.postgresql import TSMULTIRANGE from sqlalchemy.orm import DeclarativeBase from sqlalchemy.orm import Mapped from sqlalchemy.orm import mapped_column class Base(DeclarativeBase): pass class EventCalendar(Base): __tablename__ = "event_calendar" id: Mapped[int] = mapped_column(primary_key=True) event_name: Mapped[str] added: Mapped[datetime] in_session_periods: Mapped[List[Range[datetime]]] = mapped_column(TSMULTIRANGE)
插入和选择记录的示例:
from sqlalchemy import create_engine from sqlalchemy import select from sqlalchemy.orm import Session engine = create_engine("postgresql+psycopg://scott:tiger@pg14/test") Base.metadata.create_all(engine) with Session(engine) as session: calendar = EventCalendar( event_name="SQLAlchemy Tutorial Sessions", in_session_periods=[ Range(datetime(2013, 3, 23), datetime(2013, 3, 25)), Range(datetime(2013, 4, 12), datetime(2013, 4, 15)), Range(datetime(2013, 5, 9), datetime(2013, 5, 12)), ], ) session.add(calendar) session.commit() for multirange in session.scalars(select(EventCalendar.in_session_periods)): for range_ in multirange: print(f"Start: {range_.lower} End: {range_.upper}")
注意
在上面的示例中,ORM 处理的 Range
类型列表不会自动检测到特定列表值的就地更改;要使用 ORM 更新列表值,要么重新分配一个新列表给属性,要么使用 MutableList
类型修饰符。请参阅 Mutation Tracking 部分了解背景信息。
使用 MultiRange 序列推断多范围类型
当使用多范围作为字面值而不指定类型时,可以使用实用程序 MultiRange
序列:
from sqlalchemy import literal from sqlalchemy.dialects.postgresql import MultiRange with Session(engine) as session: stmt = select(EventCalendar).where( EventCalendar.added.op("<@")( MultiRange( [ Range(datetime(2023, 1, 1), datetime(2013, 3, 31)), Range(datetime(2023, 7, 1), datetime(2013, 9, 30)), ] ) ) ) in_range = session.execute(stmt).all() with engine.connect() as conn: row = conn.scalar(select(literal(MultiRange([Range(2, 4)])))) print(f"{row.lower} -> {row.upper}")
使用简单的 list
而不是 MultiRange
将需要手动设置字面值的类型为适当的多范围类型。
新版本 2.0.26 中:添加了 MultiRange
序列。
可用的多范围数据类型如下:
INT4MULTIRANGE
INT8MULTIRANGE
NUMMULTIRANGE
DATEMULTIRANGE
TSMULTIRANGE
TSTZMULTIRANGE
## 网络数据类型
包含的网络数据类型为 INET
、CIDR
、MACADDR
。
对于 INET
和 CIDR
数据类型,可条件性地支持这些数据类型发送和检索 Python ipaddress
对象,包括 ipaddress.IPv4Network
、ipaddress.IPv6Network
、ipaddress.IPv4Address
、ipaddress.IPv6Address
。这种支持目前是 DBAPI 本身的默认行为,并且因 DBAPI 而异。SQLAlchemy 尚未实现自己的网络地址转换逻辑。
- psycopg 和 asyncpg 完全支持这些数据类型;默认情况下,
ipaddress
家族的对象将在行中返回。 - psycopg2 方言仅发送和接收字符串。
- pg8000 方言支持
ipaddress.IPv4Address
和ipaddress.IPv6Address
对象用于INET
数据类型,但对于CIDR
类型则使用字符串。
要将所有上述 DBAPI 规范化为仅返回字符串,请使用 native_inet_types
参数,传递值 False
:
e = create_engine( "postgresql+psycopg://scott:tiger@host/dbname", native_inet_types=False )
使用上述参数,psycopg
、asyncpg
和 pg8000
方言将禁用这些类型的 DBAPI 适配,并仅返回字符串,与旧版 psycopg2
方言的行为相匹配。
参数也可以设置为 True
,这将导致对于那些不支持或尚未完全支持将行转换为 Python ipaddress
数据类型的后端(目前为 psycopg2 和 pg8000),引发 NotImplementedError
:
2.0.18 版本中的新功能:- 添加了 native_inet_types
参数。
SqlAlchemy 2.0 中文文档(四十八)(2)https://developer.aliyun.com/article/1563008