版本:1.1.0b2 |发布日期:2016年7月1日

SQLAlchemy 1.1文档

自定义SQL构造和编译扩展

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

概要¶ T0>

用法涉及创建一个或多个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

Above, MyColumn extends ColumnClause, the base expression element for named column objects. 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)

当使用任何postgresql方言时,第二个visit_alter_table将被调用。

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

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),
        compiler.process(element.select)
    )

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构造可能希望启用“autocommit”。有关此步骤,请参阅Enabling Autocommit on a Construct

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

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

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

上面,我们在SQLCompiler.process()(即literal_binds)标志调用的过程步骤中添加了一个额外的标志。This indicates that any SQL expression which refers to a BindParameter object or other “literal” object such as those which refer to strings or integers should be rendered in-place, rather than being referred to as a bound parameter; when emitting DDL, bound parameters are typically not supported.

在构造上启用自动提交

当被要求在没有用户定义事务的情况下执行构造时,回顾Understanding Autocommit时,Engine检测给定构造是否表示DML或DDL,也就是说,数据修改或数据定义语句需要(或者可能需要,在DDL的情况下)由DBAPI生成的事务被提交(回想起,无论SQLAlchemy做什么,DBAPI总是有一个事务正在进行)。通过检查结构上的“自动提交”执行选项实际上可以完成检查。在构建像INSERT派生,新的DDL类型或可能改变数据的存储过程的构造时,需要设置“autocommit”选项以使语句在“无连接”执行时运行(如Connectionless Execution, Implicit Execution)。

目前,一个快速的方法是将Executable子类化,然后将“autocommit”标志添加到_execution_options字典中(注意这是一个“冻结” union()方法):

from sqlalchemy.sql.expression import Executable, ClauseElement

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

More succinctly, if the construct is truly similar to an INSERT, UPDATE, or DELETE, UpdateBase can be used, which already is a subclass of Executable, ClauseElement and includes the autocommit flag:

from sqlalchemy.sql.expression import UpdateBase

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

子类DDLElement的DDL元素已经打开了“autocommit”标志。

更改现有构造的默认编译

编译器扩展也适用于现有的构造。当覆盖内置的SQL构造的编译时,@compiles装饰器将在适当的类上调用(确保使用该类,即InsertSelect),而不是创建函数,如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 classes want to have a type member which is expression’s return type. 这可以在构造函数的实例级别建立,也可以在类级别建立,如果它通常是常量:

    class timestamp(ColumnElement):
        type = TIMESTAMP()
  • FunctionElement - 这是ColumnElement和“from子句”类似对象的混合体,并且表示SQL函数或存储过程类型的调用。Since most databases support statements along the line of “SELECT FROM FunctionElement adds in the ability to be used in the FROM clause of a select() construct:

    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)
    
    @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)
  • DDLElement - 所有DDL表达式的根,例如CREATE TABLE,ALTER TABLE等。DDLElement子类的编译由DDLCompiler而不是SQLCompiler发布。DDLElement also features Table and MetaData event hooks via the execute_at() method, allowing the construct to be invoked during CREATE TABLE and DROP TABLE sequences.

  • Executable - This is a mixin which should be used with any expression class that represents a “standalone” SQL statement that can be passed directly to an execute() method. 它已经隐含在DDLElementFunctionElement中。

更多示例

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

@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
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 "CASE WHEN %s > %s THEN %s ELSE %s END" % (
        compiler.process(arg1),
        compiler.process(arg2),
        compiler.process(arg1),
        compiler.process(arg2),
    )

用法示例:

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):
    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])
)
sqlalchemy.ext.compiler.compiles(class_, *specs)

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

sqlalchemy.ext.compiler。 T0> 注销 T1> ( T2> 类_ T3> ) T4> ¶< / T5>

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