SqlAlchemy 2.0 中文文档(三十七)(2)https://developer.aliyun.com/article/1562698
为自定义构造启用缓存支持
截至版本 1.4,SQLAlchemy 包括一个 SQL 编译缓存功能,它将允许等效的 SQL 构造缓存它们的字符串形式,以及用于从语句获取结果的其他结构信息。
由于讨论的原因在对象不会生成缓存键,性能影响,这个缓存系统的实现采用了一种保守的方式来包括自定义 SQL 构造和/或子类在缓存系统中。这包括任何用户定义的 SQL 构造,包括此扩展的所有示例,默认情况下将不参与缓存,除非它们明确声明能够参与缓存。当HasCacheKey.inherit_cache
属性在特定子类的类级别上设置为True
时,将表示此类的实例可以安全地缓存,使用其直接超类的缓存键生成方案。例如,这适用于先前指示的“概要”示例:
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 编译的状态的对象示例是在编译自定义表达式结构的子元素中所示的对象;这是一个将 Table
与 Select
构造组合在一起的“INSERT FROM 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) )
“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) )
SqlAlchemy 2.0 中文文档(三十七)(4)https://developer.aliyun.com/article/1562705