Release: 1.4.25 | Release Date: September 22, 2021

SQLAlchemy 1.4 Documentation

自定义SQL构造和编译扩展

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

简介

使用涉及创建一个或多个 ClauseElement 子类和定义其编译的一个或多个可调用文件:

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

class MyColumn(ColumnClause):
    pass

@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):

    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.dialectcompiler.statement 等等 SQLCompilerDDLCompiler 两者都包括 process() 可用于编译嵌入属性的方法:

from sqlalchemy.sql.expression import Executable, ClauseElement

class InsertFromSelect(Executable, ClauseElement):
    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() 方法。

注解

以上 InsertFromSelect 构造可能希望启用“自动提交”。见 在构造上启用自动提交 对于这个步骤。

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

SQL和DDL构造都是使用不同的基本编译器编译的。- SQLCompilerDDLCompiler . 通常需要从DDL表达式中访问SQL表达式的编译规则。这个 DDLCompiler 包括访问器 sql_compiler 因此,例如下面我们在其中生成一个嵌入SQL表达式的检查约束:

@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 对象或其他“文本”对象,如引用字符串或整数的对象,应呈现 in-place ,而不是被称为绑定参数;在发出DDL时,通常不支持绑定参数。

在构造上启用自动提交

从本节召回 库级(例如仿真)自动提交Engine 当在没有用户定义的事务的情况下被要求执行一个构造时,检测给定的构造是否表示DML或DDL,即数据修改或数据定义语句,它要求(或可能要求,在DDL的情况下)提交由DBAPI生成的事务(请记住,DBAPI总是有一个事务G不管SQLAlchemy做什么,都要继续。检查这一点实际上是通过检查构造上的“autocommit”执行选项来完成的。当构建一个像insert derivation、一个新的ddl类型或者可能是一个改变数据的存储过程这样的构造时,需要设置“autocommit”选项,以便语句以“无连接”执行方式运行(如中所述 无连接执行,隐式执行

目前,一种快速的方法是子类 Executable ,然后将“自动提交”标志添加到 _execution_options 字典(注:这是一个“冻结”的字典,提供了一个生成 union() 方法:

from sqlalchemy.sql.expression import Executable, ClauseElement

class MyInsertThing(Executable, ClauseElement):
    _execution_options = \
        Executable._execution_options.union({'autocommit': True})

更简洁地说,如果构造确实类似于插入、更新或删除, UpdateBase 可以使用,它已经是 ExecutableClauseElement 包括 autocommit 旗帜:

from sqlalchemy.sql.expression import UpdateBase

class MyInsertThing(UpdateBase):
    def __init__(self, ...):
        ...

子类的DDL元素 DDLElement 已打开“自动提交”标志。

更改现有构造的默认编译

编译器扩展也适用于现有的构造。当重写内置SQL构造的编译时,@compiles修饰符将在适当的类上调用(请确保使用该类,即 InsertSelect ,而不是创建函数,例如 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 works for types, too, such as below where we implement the MS-SQL specific 'max' keyword for String/VARCHAR ::

@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的“comparison”行为。

    ColumnElement 班级希望有一个 type 是表达式的返回类型的成员。这可以在构造函数的实例级别上建立,或者在类级别上建立,如果它的一般常量为:

    class timestamp(ColumnElement):
        type = TIMESTAMP()
  • FunctionElement -这是一个混合的 ColumnElement 和“FROM子句”类似,表示SQL函数或调用的存储过程类型。因为大多数数据库都支持“select from<some function>”行中的语句。 FunctionElement 添加要在的From子句中使用的功能 select() 结构:

    from sqlalchemy.sql.expression import FunctionElement
    
    class coalesce(FunctionElement):
        name = 'coalesce'
    
    @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)
  • DDLElement -所有DDL表达式的根,如create table、alter table等编译 DDLElement 子类由 DDLCompiler 而不是 SQLCompiler . DDLElement 还特点 TableMetaData 事件挂钩通过 execute_at() 方法,允许在创建表和删除表序列期间调用构造。

  • Executable -这是一个mixin,应该与表示“独立”SQL语句的任何表达式类一起使用,该语句可以直接传递给 execute() 方法。它已经隐含在 DDLElementFunctionElement .

其他示例

“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()

@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())
)

“最大”功能

“最大的”函数被赋予任意数量的参数,并返回一个值最高的参数-相当于python的 max 功能。与仅包含两个参数的基于案例的版本相比,SQL标准版本:

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'

@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
        )

“假”表达

呈现“假”常量表达式,在没有“假”常量的平台上呈现为“0”::

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

class sql_false(expression.ColumnElement):
    pass

@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)
)
Object Name Description

compiles(class_, *specs)

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

deregister(class_)

删除与给定的 ClauseElement 类型。

function sqlalchemy.ext.compiler.compiles(class_, *specs)

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

function sqlalchemy.ext.compiler.deregister(class_)

删除与给定的 ClauseElement 类型。

Previous: SQL和通用函数 Next: 表达式序列化程序扩展