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

SqlAlchemy 2.0 中文文档（五十三）（2）https://developer.aliyun.com/article/1563173

元数据 / 模式

原文：docs.sqlalchemy.org/en/20/faq/metadata_schema.html

当我说 table.drop() / metadata.drop_all() 时，我的程序挂起了
SQLAlchemy 支持 ALTER TABLE、CREATE VIEW、CREATE TRIGGER、Schema 升级功能吗？
我如何按照它们的依赖关系对 Table 对象进行排序？
我如何将 CREATE TABLE/ DROP TABLE 输出作为字符串获取？
我如何子类化 Table/Column 以提供某些行为/配置？

当我说 `table.drop()` / `metadata.drop_all()` 时，我的程序挂起了

这通常对应两种情况：1. 使用 PostgreSQL，它对表锁非常严格，2. 您仍然打开了一个包含对表的锁定的连接，并且与用于 DROP 语句的连接不同。这是模式的最简化版本：

connection = engine.connect()
result = connection.execute(mytable.select())
mytable.drop(engine)

上述，连接池连接仍然被检出；此外，上述结果对象还保持对此连接的链接。如果使用“隐式执行”，结果将保持此连接打开，直到结果对象关闭或所有行都被耗尽。

调用 mytable.drop(engine) 试图在从 Engine 获取的第二连接上发出 DROP TABLE 操作，这将会被锁定。

解决方法是在发出 DROP TABLE 前关闭所有连接：

connection = engine.connect()
result = connection.execute(mytable.select())
# fully read result sets
result.fetchall()
# close connections
connection.close()
# now locks are removed
mytable.drop(engine)

SQLAlchemy 支持 ALTER TABLE、CREATE VIEW、CREATE TRIGGER、Schema 升级功能吗？

SQLAlchemy 直接不支持通用 ALTER。对于特殊的 ad-hoc 基础 DDL，可以使用 DDL 和相关构造。有关此主题的讨论，请参阅自定义 DDL。

更全面的选择是使用模式迁移工具，例如 Alembic 或 SQLAlchemy-Migrate；有关此问题的讨论，请参阅通过迁移修改数据库对象。

我如何按照它们的依赖关系对 Table 对象进行排序？

这可以通过 MetaData.sorted_tables 函数获得：

metadata_obj = MetaData()
# ... add Table objects to metadata
ti = metadata_obj.sorted_tables
for t in ti:
    print(t)

如何将 CREATE TABLE/ DROP TABLE 输出作为字符串获取？

现代的 SQLAlchemy 具有表示 DDL 操作的从句构造。这些可以像任何其他 SQL 表达式一样渲染为字符串：

from sqlalchemy.schema import CreateTable
print(CreateTable(mytable))

要获得特定于某个引擎的字符串：

print(CreateTable(mytable).compile(engine))

还有一种特殊形式的Engine可通过create_mock_engine()访问，允许将整个元数据创建序列转储为字符串，使用以下方法：

from sqlalchemy import create_mock_engine
def dump(sql, *multiparams, **params):
    print(sql.compile(dialect=engine.dialect))
engine = create_mock_engine("postgresql+psycopg2://", dump)
metadata_obj.create_all(engine, checkfirst=False)

Alembic 工具还支持一种“离线”SQL 生成模式，将数据库迁移呈现为 SQL 脚本。

如何对表/列进行子类化以提供特定的行为/配置？

Table 和 Column 不是直接子类化的良好目标。但是，可以使用创建函数来获取构造时的行为，并使用附加事件来处理与模式对象之间的链接，例如约束约定或命名约定。可以在命名约定中看到许多这些技术的示例。

当我说`table.drop()` / `metadata.drop_all()`时，我的程序挂起了。

这通常对应于两个条件：1. 使用 PostgreSQL，它对表锁非常严格，2. 你有一个仍然打开的连接，其中包含对表的锁，并且与用于 DROP 语句的连接不同。以下是该模式的最简版本：

connection = engine.connect()
result = connection.execute(mytable.select())
mytable.drop(engine)

上面，连接池连接仍然被检出；此外，上述结果对象还维护对此连接的链接。如果使用“隐式执行”，结果将保持此连接打开，直到关闭结果对象或耗尽所有行。

对mytable.drop(engine)的调用尝试在从Engine获取的第二个连接上发出 DROP TABLE，这会导致锁定。

解决方案是在发出 DROP TABLE 前关闭所有连接：

connection = engine.connect()
result = connection.execute(mytable.select())
# fully read result sets
result.fetchall()
# close connections
connection.close()
# now locks are removed
mytable.drop(engine)

SQLAlchemy 支持 ALTER TABLE、CREATE VIEW、CREATE TRIGGER、Schema 升级功能吗？

SQLAlchemy 并不直接支持一般的 ALTER。对于特殊的按需 DDL，可以使用DDL和相关构造。有关此主题的讨论，请参阅自定义 DDL。

更全面的选项是使用模式迁移工具，例如 Alembic 或 SQLAlchemy-Migrate；请参阅通过迁移更改数据库对象以讨论此问题。

如何按其依赖顺序对 Table 对象进行排序？

可通过MetaData.sorted_tables函数进行访问：

metadata_obj = MetaData()
# ... add Table objects to metadata
ti = metadata_obj.sorted_tables
for t in ti:
    print(t)

如何将 CREATE TABLE/DROP TABLE 输出为字符串？

现代的 SQLAlchemy 有表示 DDL 操作的子句构造。这些可以像任何其他 SQL 表达式一样渲染为字符串：

from sqlalchemy.schema import CreateTable
print(CreateTable(mytable))

要获取特定引擎的字符串：

print(CreateTable(mytable).compile(engine))

还有一种特殊形式的 Engine 可以通过 create_mock_engine() 获得，它允许将整个元数据创建序列转储为字符串，使用以下方法：

from sqlalchemy import create_mock_engine
def dump(sql, *multiparams, **params):
    print(sql.compile(dialect=engine.dialect))
engine = create_mock_engine("postgresql+psycopg2://", dump)
metadata_obj.create_all(engine, checkfirst=False)

Alembic 工具还支持一种“离线”SQL 生成模式，将数据库迁移呈现为 SQL 脚本。

我如何子类化 Table/Column 以提供某些行为/配置？

Table 和 Column 不适合直接进行子类化。但是，可以使用创建函数来获得在构造时的行为，以及使用附加事件来处理模式对象之间的链接行为，比如约束惯例或命名惯例。许多这些技术的示例可以在命名约定中看到。

SQL 表达式

原文：docs.sqlalchemy.org/en/20/faq/sqlexpressions.html

如何将 SQL 表达式呈现为字符串，可能包含内联的绑定参数？

针对特定数据库进行字符串化
内联呈现绑定参数
将“POSTCOMPILE”参数呈现为绑定参数

在字符串化 SQL 语句时为什么百分号会被双倍显示？
我正在使用 op() 生成自定义运算符，但我的括号没有正确显示

为什么括号规则是这样的？

如何将 SQL 表达式呈现为字符串，可能包含内联的绑定参数？

SQLAlchemy Core 语句对象或表达式片段的“字符串化”，以及 ORM Query 对象，在大多数简单情况下都可以简单地使用 str() 内置函数来实现，如下所示，当与 print 函数一起使用时（请注意 Python print 函数如果不显式使用 str()，也会自动调用它）：

>>> from sqlalchemy import table, column, select
>>> t = table("my_table", column("x"))
>>> statement = select(t)
>>> print(str(statement))
SELECT  my_table.x
FROM  my_table

str() 内置函数或等效函数，可在 ORM Query 对象上调用，也可在诸如 select()、insert() 等语句上调用，还可在任何表达式片段上调用，例如：

>>> from sqlalchemy import column
>>> print(column("x") == "some value")
x  =  :x_1

针对特定数据库进行字符串化

当我们要将语句或片段字符串化时，如果包含具有特定于数据库的字符串格式的元素，或者包含仅在某种类型的数据库中可用的元素，则会出现复杂情况。在这些情况下，我们可能会得到一个不符合我们目标数据库正确语法的字符串化语句，或者操作可能会引发一个UnsupportedCompilationError异常。在这些情况下，有必要使用ClauseElement.compile()方法将语句字符串化，同时传递一个代表目标数据库的Engine或Dialect对象。例如，如果我们有一个 MySQL 数据库引擎，我们可以按照 MySQL 方言字符串化一个语句：

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://scott:tiger@localhost/test")
print(statement.compile(engine))

更直接地，不需要构建一个Engine对象，我们可以直接实例化一个Dialect对象，如下所示，我们使用一个 PostgreSQL 方言：

from sqlalchemy.dialects import postgresql
print(statement.compile(dialect=postgresql.dialect()))

请注意，任何方言都可以使用create_engine()本身组装，使用一个虚拟 URL，然后访问Engine.dialect属性，比如如果我们想要一个 psycopg2 的方言对象：

e = create_engine("postgresql+psycopg2://")
psycopg2_dialect = e.dialect

给定一个 ORM Query对象时，为了访问ClauseElement.compile()方法，我们只需要首先访问Query.statement访问器：

statement = query.statement
print(statement.compile(someengine))

内联渲染绑定参数

警告

永远不要使用这些技术处理来自不受信任输入的字符串内容，比如来自 Web 表单或其他用户输入应用程序。SQLAlchemy 将 Python 值强制转换为直接 SQL 字符串值的功能不安全，并且不验证传递的数据类型。在针对关系数据库编程调用非 DDL SQL 语句时，始终使用绑定参数。

上述形式将渲染 SQL 语句，因为它被传递到 Python DBAPI，其中包括绑定参数不会内联渲染。SQLAlchemy 通常不会对绑定参数进行字符串化处理，因为这由 Python DBAPI 适当处理，更不用说绕过绑定参数可能是现代 Web 应用中被广泛利用的安全漏洞之一了。SQLAlchemy 在某些情况下有限的能力执行此字符串化，例如发出 DDL 时。为了访问此功能，可以使用传递给compile_kwargs的literal_binds标志：

from sqlalchemy.sql import table, column, select
t = table("t", column("x"))
s = select(t).where(t.c.x == 5)
# **do not use** with untrusted input!!!
print(s.compile(compile_kwargs={"literal_binds": True}))
# to render for a specific dialect
print(s.compile(dialect=dialect, compile_kwargs={"literal_binds": True}))
# or if you have an Engine, pass as first argument
print(s.compile(some_engine, compile_kwargs={"literal_binds": True}))

此功能主要用于日志记录或调试目的，其中获得查询的原始 sql 字符串可能会很有用。

上述方法的注意事项是，它仅支持基本类型，如整数和字符串，而且如果直接使用未设置预设值的bindparam()，它也无法对其进行字符串化处理。下面详细介绍了无条件对所有参数进行字符串化的方法。

提示

SQLAlchemy 不支持对所有数据类型进行完全字符串化的原因有三个：

当正常使用 DBAPI 时，该功能已被当前 DBAPI 支持。SQLAlchemy 项目不能被要求为所有后端的所有数据类型重复这种功能，因为这是多余的工作，还带来了重大的测试和持续支持开销。
对于特定数据库的绑定参数进行字符串化建议一种实际上将这些完全字符串化的语句传递给数据库以进行执行的用法。这是不必要和不安全的，SQLAlchemy 不希望以任何方式鼓励这种用法。
渲染字面值的区域是最有可能报告安全问题的地方。SQLAlchemy 尽量将安全参数字符串化的区域留给 DBAPI 驱动程序，这样每个 DBAPI 的具体细节可以得到适当和安全地处理。

由于 SQLAlchemy 故意不支持对所有数据类型的完全字符串化，因此在特定调试场景下执行此操作的技术包括以下内容。作为示例，我们将使用 PostgreSQL 的UUID数据类型：

import uuid
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import Integer
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class A(Base):
    __tablename__ = "a"
    id = Column(Integer, primary_key=True)
    data = Column(UUID)
stmt = select(A).where(A.data == uuid.uuid4())

给定上述模型和语句，将比较一列与单个 UUID 值，将此语句与内联值一起进行字符串化的选项包括：

一些 DBAPI，如 psycopg2，支持像mogrify()这样的辅助函数，提供对它们的字面渲染功能的访问。要使用此类功能，请渲染 SQL 字符串而不使用literal_binds，并通过SQLCompiler.params访问器分别传递参数：

e = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
with e.connect() as conn:
    cursor = conn.connection.cursor()
    compiled = stmt.compile(e)
    print(cursor.mogrify(str(compiled), compiled.params))

上述代码将生成 psycopg2 的原始字节串：

b"SELECT a.id, a.data \nFROM a \nWHERE a.data = 'a511b0fc-76da-4c47-a4b4-716a8189b7ac'::uuid"

直接将SQLCompiler.params渲染到语句中，使用目标 DBAPI 的适当paramstyle。例如，psycopg2 DBAPI 使用命名的pyformat样式。render_postcompile的含义将在下一节中讨论。警告，这是不安全的，请勿使用不受信任的输入：

e = create_engine("postgresql+psycopg2://")
# will use pyformat style, i.e. %(paramname)s for param
compiled = stmt.compile(e, compile_kwargs={"render_postcompile": True})
print(str(compiled) % compiled.params)

这将产生一个非工作的字符串，但仍然适合调试：

SELECT  a.id,  a.data
FROM  a
WHERE  a.data  =  9eec1209-50b4-4253-b74b-f82461ed80c1

另一个示例使用位置参数风格，如qmark，我们可以使用SQLCompiler.positiontup集合与SQLCompiler.params一起编译我们上面的语句，以便按其位置顺序检索语句的参数：

import re
e = create_engine("sqlite+pysqlite://")
# will use qmark style, i.e. ? for param
compiled = stmt.compile(e, compile_kwargs={"render_postcompile": True})
# params in positional order
params = (repr(compiled.params[name]) for name in compiled.positiontup)
print(re.sub(r"\?", lambda m: next(params), str(compiled)))

上述片段打印：

SELECT  a.id,  a.data
FROM  a
WHERE  a.data  =  UUID('1bd70375-db17-4d8c-94f1-fc2ef3aada26')

使用自定义 SQL 构造和编译扩展扩展，在用户定义的标志存在时以自定义方式渲染BindParameter对象。这个标志通过compile_kwargs字典发送，就像任何其他标志一样：

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import BindParameter
@compiles(BindParameter)
def _render_literal_bindparam(element, compiler, use_my_literal_recipe=False, **kw):
    if not use_my_literal_recipe:
        # use normal bindparam processing
        return compiler.visit_bindparam(element, **kw)
    # if use_my_literal_recipe was passed to compiler_kwargs,
    # render the value directly
    return repr(element.value)
e = create_engine("postgresql+psycopg2://")
print(stmt.compile(e, compile_kwargs={"use_my_literal_recipe": True}))

上述配方将打印：

SELECT  a.id,  a.data
FROM  a
WHERE  a.data  =  UUID('47b154cd-36b2-42ae-9718-888629ab9857')

对于内置于模型或语句中的特定类型的字符串化，可以使用TypeDecorator类使用TypeDecorator.process_literal_param()方法来提供任何数据类型的自定义字符串化：

from sqlalchemy import TypeDecorator
class UUIDStringify(TypeDecorator):
    impl = UUID
    def process_literal_param(self, value, dialect):
        return repr(value)

上述数据类型需要在模型内部明确使用，或者在语句内部使用type_coerce()，例如

from sqlalchemy import type_coerce
stmt = select(A).where(type_coerce(A.data, UUIDStringify) == uuid.uuid4())
print(stmt.compile(e, compile_kwargs={"literal_binds": True}))

再次打印相同形式：

SELECT  a.id,  a.data
FROM  a
WHERE  a.data  =  UUID('47b154cd-36b2-42ae-9718-888629ab9857')

SqlAlchemy 2.0 中文文档（五十三）（4）https://developer.aliyun.com/article/1563175

SqlAlchemy 2.0 中文文档（五十三）（3）

元数据 / 模式

当我说 `table.drop()` / `metadata.drop_all()` 时，我的程序挂起了

SQLAlchemy 支持 ALTER TABLE、CREATE VIEW、CREATE TRIGGER、Schema 升级功能吗？

我如何按照它们的依赖关系对 Table 对象进行排序？

如何将 CREATE TABLE/ DROP TABLE 输出作为字符串获取？

如何对表/列进行子类化以提供特定的行为/配置？

当我说`table.drop()` / `metadata.drop_all()`时，我的程序挂起了。

SQLAlchemy 支持 ALTER TABLE、CREATE VIEW、CREATE TRIGGER、Schema 升级功能吗？

如何按其依赖顺序对 Table 对象进行排序？

如何将 CREATE TABLE/DROP TABLE 输出为字符串？

我如何子类化 Table/Column 以提供某些行为/配置？

SQL 表达式

如何将 SQL 表达式呈现为字符串，可能包含内联的绑定参数？

针对特定数据库进行字符串化

内联渲染绑定参数

热门文章

最新文章

相关电子书

SqlAlchemy 2.0 中文文档（五十三）（3）

元数据 / 模式

当我说 table.drop() / metadata.drop_all() 时，我的程序挂起了

SQLAlchemy 支持 ALTER TABLE、CREATE VIEW、CREATE TRIGGER、Schema 升级功能吗？

我如何按照它们的依赖关系对 Table 对象进行排序？

如何将 CREATE TABLE/ DROP TABLE 输出作为字符串获取？

如何对表/列进行子类化以提供特定的行为/配置？

当我说table.drop() / metadata.drop_all()时，我的程序挂起了。

SQLAlchemy 支持 ALTER TABLE、CREATE VIEW、CREATE TRIGGER、Schema 升级功能吗？

如何按其依赖顺序对 Table 对象进行排序？

如何将 CREATE TABLE/DROP TABLE 输出为字符串？

我如何子类化 Table/Column 以提供某些行为/配置？

SQL 表达式

如何将 SQL 表达式呈现为字符串，可能包含内联的绑定参数？

针对特定数据库进行字符串化

内联渲染绑定参数

热门文章

最新文章

相关电子书

当我说 `table.drop()` / `metadata.drop_all()` 时，我的程序挂起了

当我说`table.drop()` / `metadata.drop_all()`时，我的程序挂起了。