自定义 SQL 构造和编译扩展¶

提供用于创建自定义 ClauseElements 和编译器的 API。

剧情简介¶

使用涉及创建一个或多个 ClauseElement 子类和一个或多个定义其编译的可调用对象：

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import ColumnClause


class MyColumn(ColumnClause):
    inherit_cache = True


@compiles(MyColumn)
def compile_mycolumn(element, compiler, **kw):
    return "[%s]" % element.name

在上面，MyColumn 扩展了 ColumnClause，这是命名列对象的基表达式元素。编译 装饰器向 MyColumn 类注册自身，以便在将对象编译为字符串时调用它：

from sqlalchemy import select

s = select(MyColumn("x"), MyColumn("y"))
print(str(s))

生产：

SELECT [x], [y]

Dialect 特定的编译规则¶

编译器也可以特定于方言。将为正在使用的方言调用适当的编译器：

from sqlalchemy.schema import DDLElement


class AlterColumn(DDLElement):
    inherit_cache = False

    def __init__(self, column, cmd):
        self.column = column
        self.cmd = cmd


@compiles(AlterColumn)
def visit_alter_column(element, compiler, **kw):
    return "ALTER COLUMN %s ..." % element.column.name


@compiles(AlterColumn, "postgresql")
def visit_alter_column(element, compiler, **kw):
    return "ALTER TABLE %s ALTER COLUMN %s ..." % (
        element.table.name,
        element.column.name,
    )

第二个 visit_alter_table 将在任何 postgresql 方言。

编译自定义表达式结构的子元素¶

compiler 参数是正在使用的已编译对象。可以检查此对象以获取有关正在进行的编译的任何信息，包括 compiler.dialect、compiler.statement 等。这 SQLCompiler 和 DDLCompiler 都包含一个 process（） 方法，可用于编译嵌入属性：

from sqlalchemy.sql.expression import Executable, ClauseElement


class InsertFromSelect(Executable, ClauseElement):
    inherit_cache = False

    def __init__(self, table, select):
        self.table = table
        self.select = select


@compiles(InsertFromSelect)
def visit_insert_from_select(element, compiler, **kw):
    return "INSERT INTO %s (%s)" % (
        compiler.process(element.table, asfrom=True, **kw),
        compiler.process(element.select, **kw),
    )


insert = InsertFromSelect(t1, select(t1).where(t1.c.x > 5))
print(insert)

生成（为便于阅读而格式化）：

INSERT INTO mytable (
    SELECT mytable.x, mytable.y, mytable.z
    FROM mytable
    WHERE mytable.x > :x_1
)

注意

上面的 InsertFromSelect 构造只是一个示例，这个实际的功能已经可以使用 Insert.from_select（） 方法。

SQL 和 DDL 编译器之间的交叉编译¶

SQL 和 DDL 结构分别使用不同的基本编译器进行编译 - SQLCompiler 和 DDLCompiler。一个常见的需求是访问 DDL 表达式中 SQL 表达式的编译规则。这 因此，DDLCompiler 包含一个访问器sql_compiler，例如下面我们生成嵌入 SQL 表达式的 CHECK 约束：

@compiles(MyConstraint)
def compile_my_constraint(constraint, ddlcompiler, **kw):
    kw["literal_binds"] = True
    return "CONSTRAINT %s CHECK (%s)" % (
        constraint.name,
        ddlcompiler.sql_compiler.process(constraint.expression, **kw),
    )

在上面，我们向流程步骤添加了一个额外的标志，由 SQLCompiler.process（）来执行，这是 literal_binds 标志。这表示引用 BindParameter 的任何 SQL 表达式 object 或其他 “literal” 对象，例如引用字符串或整数应该就地呈现，而不是称为绑定参数;发出 DDL 时，通常不支持绑定参数。

更改现有结构的默认编译¶

编译器扩展同样适用于现有构造。当覆盖内置 SQL 结构的编译时，将在适当的类上调用 @compiles 装饰器（请务必使用该类，即 Insert 或 Select，而不是 insert（） 或 select（） 等创建函数）。

在新的编译函数中，要获得“原始”编译例程，请使用适当的 visit_XXX 方法 - 这是因为 compiler.process（）将调用覆盖例程并导致无限循环。例如，要将 “prefix” 添加到所有 insert 语句中：

from sqlalchemy.sql.expression import Insert


@compiles(Insert)
def prefix_inserts(insert, compiler, **kw):
    return compiler.visit_insert(insert.prefix_with("some prefix"), **kw)

上述编译器在编译时将为所有 INSERT 语句加上 “some prefix” 前缀。

更改类型的编译¶

compiler 也适用于类型，例如下面我们为 String/VARCHAR 实现特定于 MS-SQL 的 'max' 关键字：

@compiles(String, "mssql")
@compiles(VARCHAR, "mssql")
def compile_varchar(element, compiler, **kw):
    if element.length == "max":
        return "VARCHAR('max')"
    else:
        return compiler.visit_VARCHAR(element, **kw)


foo = Table("foo", metadata, Column("data", VARCHAR("max")))

子类化指南¶

使用编译器扩展的很大一部分是对 SQLAlchemy 表达式构造进行子类化。为了简化此作，表达式和架构包具有一组用于常见任务的“基础”。概要如下：

ClauseElement - 这是根表达式类。任何 SQL 表达式都可以从此基础派生，并且可能是较长结构（如专用 INSERT 语句）的最佳选择。
ColumnElement - 所有“列状”元素的根。您放置在 SELECT 语句的 “columns” 子句中的任何内容（以及 order by 和 group by）都可以从中派生 - 该对象将自动具有 Python 的“比较”行为。

ColumnElement 类希望具有 type 成员，该成员是 expression 的返回类型。这可以在构造函数的实例级别建立，如果它通常是常数，则可以在类级别建立：
```
class timestamp(ColumnElement):
    type = TIMESTAMP()
    inherit_cache = True
```

FunctionElement - 这是 ColumnElement 和类似 object 的 “from 子句” ，并表示一个 SQL 函数或存储过程类型的调用。由于大多数数据库都支持沿 “SELECT FROM ” 行的语句 FunctionElement 添加了在 select（） 结构中：

from sqlalchemy.sql.expression import FunctionElement


class coalesce(FunctionElement):
    name = "coalesce"
    inherit_cache = True


@compiles(coalesce)
def compile(element, compiler, **kw):
    return "coalesce(%s)" % compiler.process(element.clauses, **kw)


@compiles(coalesce, "oracle")
def compile(element, compiler, **kw):
    if len(element.clauses) > 2:
        raise TypeError(
            "coalesce only supports two arguments on " "Oracle Database"
        )
    return "nvl(%s)" % compiler.process(element.clauses, **kw)

ExecutableDDLElement - 所有 DDL 表达式的根，如 CREATE TABLE、ALTER TABLE 等。编译 ExecutableDDLElement 子类由 DDLCompiler 而不是 SQLCompiler 的 SQL 编译器。 ExecutableDDLElement 还可以与 DDLEvents.before_create（）和 DDLEvents.after_create（） 中，允许在 CREATE TABLE 和 DROP TABLE 序列期间自动调用构造。

另请参阅

自定义 DDL - 包含关联示例 DDL 对象（本身是 ExecutableDDLElement 实例）与 DDLEvents 事件钩子一起使用。
可执行 - 这是一个 mixin，应该与任何表示可以直接传递给 execute（） 方法的 “独立” SQL 语句的表达式类一起使用。它已经隐含在 DDLElement 和 FunctionElement 中。

上述大多数构造也响应 SQL 语句缓存。子类化构造需要定义对象的缓存行为，这通常意味着将标志 inherit_cache 设置为 False 或 True。请参阅下一部分：为自定义构造启用缓存支持作为背景。

启用对自定义结构的缓存支持¶

从 1.4 版本开始，SQLAlchemy 包含一个 SQL 编译缓存工具，它将允许等效的 SQL 结构缓存其字符串化形式，以及用于从语句中获取结果的其他结构信息。

由于 Object 不会生成缓存键性能影响中讨论的原因，此缓存系统的实现采用保守的方法，在缓存系统中包含自定义 SQL 构造和/或子类。这包括任何用户定义的 SQL 结构（包括此扩展的所有示例）默认情况下都不会参与缓存，除非它们肯定地断言它们能够这样做。HasCacheKey.inherit_cache 属性时，在特定子类的类级别设置为 True 时，将指示可以使用直接超类的缓存键生成方案安全地缓存此类的实例。例如，这适用于前面提到的 “synopsis” 示例：

class MyColumn(ColumnClause):
    inherit_cache = True


@compiles(MyColumn)
def compile_mycolumn(element, compiler, **kw):
    return "[%s]" % element.name

在上面，MyColumn 类不包含任何影响其 SQL 编译的新状态;MyColumn 实例的缓存键将使用 ColumnClause 超类的缓存键，这意味着它将考虑对象的类（MyColumn）、对象的字符串名称和数据类型：

>>> MyColumn("some_name", String())._generate_cache_key()
CacheKey(
    key=('0', <class '__main__.MyColumn'>,
    'name', 'some_name',
    'type', (<class 'sqlalchemy.sql.sqltypes.String'>,
             ('length', None), ('collation', None))
), bindparams=[])

对于可能被广泛用作许多较大的语句，例如 Column 子类和自定义 SQL 数据类型，因此尽可能多地启用缓存非常重要，否则这可能会对性能产生负面影响。

包含影响其 SQL 编译的状态的对象示例是编译自定义表达式构造的子元素中说明的示例; 这是一个 “INSERT FROM SELECT” 结构，它将 Table 以及 Select 构造，每个构造都独立影响构造的 SQL 字符串生成。对于此类，该示例说明了它根本不参与缓存：

class InsertFromSelect(Executable, ClauseElement):
    inherit_cache = False

    def __init__(self, table, select):
        self.table = table
        self.select = select


@compiles(InsertFromSelect)
def visit_insert_from_select(element, compiler, **kw):
    return "INSERT INTO %s (%s)" % (
        compiler.process(element.table, asfrom=True, **kw),
        compiler.process(element.select, **kw),
    )

虽然也可以使上述 InsertFromSelect 生成一个缓存键，该键由 Table 和 一起选择组件，此 API 目前尚未完全公开。但是，对于仅由自身用于特定作的 “INSERT FROM SELECT” 构造，缓存并不像上一个示例中那样重要。

对于相对隔离使用且通常独立缓存，例如自定义 DML 构造（如“INSERT FROM SELECT”），缓存通常不太重要，因为缺少此类构造的缓存只会对该特定作产生局部影响。

更多示例¶

“UTC 时间戳” 函数¶

与 “CURRENT_TIMESTAMP” 类似但应用适当转换的函数，以便时间采用 UTC 时间。时间戳最好以 UTC 格式存储在关系数据库中，而不使用时区。UTC 的调用，这样您的数据库就不会认为夏令时结束时的时间已经倒退，没有时区，因为时区就像字符编码 - 它们最好只应用于应用程序的端点（即，在用户输入时转换为 UTC，在显示时重新应用所需的时区）。

对于 PostgreSQL 和 Microsoft SQL Server：

from sqlalchemy.sql import expression
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import DateTime


class utcnow(expression.FunctionElement):
    type = DateTime()
    inherit_cache = True


@compiles(utcnow, "postgresql")
def pg_utcnow(element, compiler, **kw):
    return "TIMEZONE('utc', CURRENT_TIMESTAMP)"


@compiles(utcnow, "mssql")
def ms_utcnow(element, compiler, **kw):
    return "GETUTCDATE()"

用法示例：

from sqlalchemy import Table, Column, Integer, String, DateTime, MetaData

metadata = MetaData()
event = Table(
    "event",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("description", String(50), nullable=False),
    Column("timestamp", DateTime, server_default=utcnow()),
)

“GREATEST” 函数¶

“GREATEST” 函数被赋予任意数量的参数，并返回具有最高值的参数 - 它相当于 Python 的最大值 功能。 SQL 标准版本与基于 CASE 的版本，后者仅容纳两个参数：

from sqlalchemy.sql import expression, case
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import Numeric


class greatest(expression.FunctionElement):
    type = Numeric()
    name = "greatest"
    inherit_cache = True


@compiles(greatest)
def default_greatest(element, compiler, **kw):
    return compiler.visit_function(element)


@compiles(greatest, "sqlite")
@compiles(greatest, "mssql")
@compiles(greatest, "oracle")
def case_greatest(element, compiler, **kw):
    arg1, arg2 = list(element.clauses)
    return compiler.process(case((arg1 > arg2, arg1), else_=arg2), **kw)

用法示例：

Session.query(Account).filter(
    greatest(Account.checking_balance, Account.savings_balance) > 10000
)

“false” 表达式¶

渲染一个 “false” 常量表达式，在没有 “false” 常量的平台上渲染为 “0”：

from sqlalchemy.sql import expression
from sqlalchemy.ext.compiler import compiles


class sql_false(expression.ColumnElement):
    inherit_cache = True


@compiles(sql_false)
def default_false(element, compiler, **kw):
    return "false"


@compiles(sql_false, "mssql")
@compiles(sql_false, "mysql")
@compiles(sql_false, "oracle")
def int_false(element, compiler, **kw):
    return "0"

用法示例：

from sqlalchemy import select, union_all

exp = union_all(
    select(users.c.name, sql_false().label("enrolled")),
    select(customers.c.name, customers.c.enrolled),
)

对象名称	描述
compiles（class_， *specs）	将函数注册为给定 `ClauseElement` 类型的编译器。
取消注册（class_）	删除与给定 `ClauseElement` 类型。

函数 sqlalchemy.ext.compiler。compiles（class_： Type[Any]， *specs： str）→ 可调用[[_F]、 _F]¶: 将函数注册为给定 ClauseElement 类型的编译器。

函数 sqlalchemy.ext.compiler。deregister（class_： Type[Any]）→ 无¶: 删除与给定 ClauseElement 类型。

SQLAlchemy 2.0 文档

SQLAlchemy 2.0 文档

SQLAlchemy 核心

自定义 SQL 构造和编译扩展¶

剧情简介¶

Dialect 特定的编译规则¶

编译自定义表达式结构的子元素¶

SQL 和 DDL 编译器之间的交叉编译¶

更改现有结构的默认编译¶

更改类型的编译¶

子类化指南¶

启用对自定义结构的缓存支持¶

更多示例¶

“UTC 时间戳” 函数¶

“GREATEST” 函数¶

“false” 表达式¶