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

SQLAlchemy 1.1文档

自定义SQL构造和编译扩展

提供用于创建自定义ClauseElement和编译器的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对象。This object can be inspected for any information about the in-progress compilation, including compiler.dialect, compiler.statement etc. 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标志。这表明引用一个BindParameter对象或其他“文字”对象的任何SQL表达式(例如那些引用字符串或整数的对象)应该呈现在in-place中,而不是被称为绑定参数;当发射DDL时,通常不支持绑定的参数。

在构造上启用自动提交

Recall from the section Understanding Autocommit that the Engine, when asked to execute a construct in the absence of a user-defined transaction, detects if the given construct represents DML or DDL, that is, a data modification or data definition statement, which requires (or may require, in the case of DDL) that the transaction generated by the DBAPI be committed (recall that DBAPI always has a transaction going on regardless of what SQLAlchemy does). 检查这实际上是通过检查结构上的“autocommit”执行选项来完成的。当像INSERT派生,新的DDL类型或可能改变数据的存储过程一样构建一个结构时,需要设置“autocommit”选项以便语句在“无连接”执行时运行(如Connectionless Execution, Implicit Execution)。

目前,一个快速的方法是将Executable子类,然后将“autocommit”标志添加到_execution_options字典中(注意这是一个“frozen” 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 - This is the root expression class. 任何SQL表达式都可以从这个基础派生,并且可能是较长的结构(如专用的INSERT语句)的最佳选择。

  • ColumnElement - The root of all “column-like” elements. 任何你在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 - The root of all DDL expressions, like CREATE TABLE, ALTER TABLE, etc. 编译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>

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