自定义 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，它是命名列对象的基表达式元素。 compiles 装饰器将自身注册到 MyColumn 类，以便在对象编译为字符串时调用它

from sqlalchemy import select

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

生成

SELECT [x], [y]

特定于方言的编译规则

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

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 参数是正在使用的 Compiled 对象。此对象可以检查有关正在进行的编译的任何信息，包括 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 表达式，例如那些引用字符串或整数的表达式，应该被就地呈现，而不是被引用为绑定参数；在发出 DDL 时，通常不支持绑定参数。

更改现有构造的默认编译

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

在新编译函数中，要获取“原始”编译例程，请使用适当的 visit_XXX 方法 - 这是因为 compiler.process() 将调用覆盖例程并导致无限循环。例如，要为所有插入语句添加“前缀”

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 成员，它是表达式的返回类型。这可以在构造函数的实例级别建立，也可以在类级别建立（如果它通常是常量）

  class timestamp(ColumnElement):
      type = TIMESTAMP()
      inherit_cache = True

• FunctionElement - 这是一个 ColumnElement 和一个类似“from 子句”的对象的混合体，它表示 SQL 函数或存储过程类型的调用。由于大多数数据库都支持类似于“SELECT FROM <some function>” 的语句，因此 FunctionElement 添加了在 select() 构造的 FROM 子句中使用的能力

  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")
      return "nvl(%s)" % compiler.process(element.clauses, **kw)

• ExecutableDDLElement - 所有 DDL 表达式的根，如 CREATE TABLE、ALTER TABLE 等等。 ExecutableDDLElement 子类的编译由 DDLCompiler 而不是 SQLCompiler 发出。 ExecutableDDLElement 也可以用作事件钩子，与 DDLEvents.before_create() 和 DDLEvents.after_create() 等事件钩子结合使用，允许构造在 CREATE TABLE 和 DROP TABLE 序列中自动调用。

  另请参阅

  自定义 DDL - 包含将 DDL 对象（本身是 ExecutableDDLElement 实例）与 DDLEvents 事件钩子关联的示例。

• Executable - 这是一个混合类，应该与任何表示可直接传递给 execute() 方法的“独立”SQL 语句的表达式类一起使用。它已经在 DDLElement 和 FunctionElement 中隐式存在。

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

为自定义构造启用缓存支持

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

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

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 和 Select 组件组成的缓存键，但此 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 的 max 函数。一个 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 类型的编译器。
deregister(class_)	删除与给定 ClauseElement 类型关联的所有自定义编译器。

function sqlalchemy.ext.compiler.compiles(class_: Type[Any], *specs: str) → Callable[[_F], _F]

将一个函数注册为给定 ClauseElement 类型的编译器。

function sqlalchemy.ext.compiler.deregister(class_: Type[Any]) → None

删除与给定 ClauseElement 类型关联的所有自定义编译器。