自定义类型

存在多种方法来重新定义现有类型的行为以及提供新的类型。

重写类型编译

一个常见的需求是强制更改类型的“字符串”版本，即在create table语句或其他SQL函数（如cast）中呈现的版本。例如，应用程序可能希望强制呈现 BINARY 适用于除一个平台外的所有平台 BLOB 待渲染。在本例中，使用现有的泛型类型 LargeBinary ，是大多数用例的首选。但是为了更准确地控制类型，每个方言的编译指令可以与任何类型相关联：

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import BINARY

@compiles(BINARY, "sqlite")
def compile_binary_sqlite(type_, compiler, **kw):
    return "BLOB"

上述代码允许使用 BINARY ，将生成字符串 BINARY 除了sqlite以外的所有后端，在这种情况下，它将生成 BLOB .

见剖面图 更改类型的编译 ，的一个子部分 自定义SQL构造和编译扩展 ，用于其他示例。

扩充现有类型

这个 TypeDecorator 允许创建将绑定参数和结果处理行为添加到现有类型对象的自定义类型。在Python中附加时使用 marshalling 需要与数据库进行数据交换。

注解

绑定和结果处理 TypeDecorator 是 此外 到已由托管类型执行的处理，该类型由SQLAlchemy根据每个DBAPI进行自定义，以执行特定于该DBAPI的处理。虽然可以通过直接子类化来替换给定类型的此处理，但在实践中并不需要它，并且SQLAlchemy不再将其作为公共用例支持。

ORM提示

这个 TypeDecorator 可用于提供在某种类型的值传入和传出数据库时转换这些值的一致方法。在使用ORM时，有一种类似的技术可以将用户数据从任意格式转换为使用 validates() 装饰工。当进入ORM模型的数据需要以某种特定于业务案例的方式规范化，而不是像数据类型那样通用时，这种技术可能更合适。

class sqlalchemy.types.TypeDecorator(*args, **kwargs)

几把钥匙 TypeDecorator 食谱如下。

将编码字符串强制为Unicode

关于 Unicode 类型是它要处理的 only 用Python unicode 对象在python端，这意味着作为绑定参数传递给它的值的形式必须是 u'some string' 如果使用的是python 2而不是3。它执行的编码/解码功能仅适用于使用中的DBAPI所需的功能，并且主要是私有实现细节。

可以安全地接收python字节串的类型的用例，即包含非ascii字符的字符串，而不是 u'' 在python 2中，可以使用 TypeDecorator 根据需要强制：

from sqlalchemy.types import TypeDecorator, Unicode

class CoerceUTF8(TypeDecorator):
    """Safely coerce Python bytestrings to Unicode
    before passing off to the database."""

    impl = Unicode

    def process_bind_param(self, value, dialect):
        if isinstance(value, str):
            value = value.decode('utf-8')
        return value

四舍五入数字

一些数据库连接器，如SQL Server连接器，如果传递的小数位数太多，则会阻塞。这里有一个菜谱可以把它们四舍五入：

from sqlalchemy.types import TypeDecorator, Numeric
from decimal import Decimal

class SafeNumeric(TypeDecorator):
    """Adds quantization to Numeric."""

    impl = Numeric

    def __init__(self, *arg, **kw):
        TypeDecorator.__init__(self, *arg, **kw)
        self.quantize_int = - self.impl.scale
        self.quantize = Decimal(10) ** self.quantize_int

    def process_bind_param(self, value, dialect):
        if isinstance(value, Decimal) and \
            value.as_tuple()[2] < self.quantize_int:
            value = value.quantize(self.quantize)
        return value

将时区感知时间戳存储为时区原始UTC

数据库中的时间戳应始终以时区无关的方式存储。对于大多数数据库，这意味着在存储时间戳之前，确保时间戳首先位于UTC时区，然后将其存储为原始时区（即，没有任何与之关联的时区；假定UTC为“隐式”时区）。另外，数据库特定的类型，如PostgreSQLs“TIMESTAMP WITH TIMEZONE”通常是因为它们更丰富的功能；但是，以普通UTC格式存储将适用于所有数据库和驱动程序。如果时区智能数据库类型不是选项或不是首选类型，则 TypeDecorator 可用于创建一个数据类型，该数据类型将时区感知时间戳转换为时区原始时间戳，然后再转换回时区。下面是Python的内置 datetime.timezone.utc 时区用于规范化和非规范化：

import datetime

class TZDateTime(TypeDecorator):
    impl = DateTime
    cache_ok = True

    def process_bind_param(self, value, dialect):
        if value is not None:
            if not value.tzinfo:
                raise TypeError("tzinfo is required")
            value = value.astimezone(datetime.timezone.utc).replace(
                tzinfo=None
            )
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = value.replace(tzinfo=datetime.timezone.utc)
        return value

后端不可知guid类型

接收并返回python uuid（）对象。在其他后端使用postgresql、char（32）时使用pg uuid类型，以字符串化十六进制格式存储它们。如果需要，可以修改为在char（16）中存储二进制：

from sqlalchemy.types import TypeDecorator, CHAR
from sqlalchemy.dialects.postgresql import UUID
import uuid

class GUID(TypeDecorator):
    """Platform-independent GUID type.

    Uses PostgreSQL's UUID type, otherwise uses
    CHAR(32), storing as stringified hex values.

    """
    impl = CHAR
    cache_ok = True

    def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(UUID())
        else:
            return dialect.type_descriptor(CHAR(32))

    def process_bind_param(self, value, dialect):
        if value is None:
            return value
        elif dialect.name == 'postgresql':
            return str(value)
        else:
            if not isinstance(value, uuid.UUID):
                return "%.32x" % uuid.UUID(value).int
            else:
                # hexstring
                return "%.32x" % value.int

    def process_result_value(self, value, dialect):
        if value is None:
            return value
        else:
            if not isinstance(value, uuid.UUID):
                value = uuid.UUID(value)
            return value

封送JSON字符串

这种类型的用途 simplejson 将python数据结构封送到JSON或从JSON传出。可以修改为使用Python的内置JSON编码器：

from sqlalchemy.types import TypeDecorator, VARCHAR
import json

class JSONEncodedDict(TypeDecorator):
    """Represents an immutable structure as a json-encoded string.

    Usage::

        JSONEncodedDict(255)

    """

    impl = VARCHAR

    cache_ok = True

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = json.dumps(value)

        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = json.loads(value)
        return value

添加可变性

默认情况下，ORM不会检测到上述类型的“可变性”，也就是说，对值的就地更改将不会被检测到，也不会被刷新。如果不采取进一步步骤，则需要在每个父对象上用新值替换现有值，以检测更改：

obj.json_value["key"] = "value"  # will *not* be detected by the ORM

obj.json_value = {"key": "value"}  # *will* be detected by the ORM

上述限制可能很好，因为许多应用程序可能不要求值在创建后发生变化。对于那些有这个需求的人，最好使用 sqlalchemy.ext.mutable 延伸。对于面向字典的JSON结构，我们可以将其应用为：

json_type = MutableDict.as_mutable(JSONEncodedDict)

class MyClass(Base):
    #  ...

    json_data = Column(json_type)

参见

突变跟踪

处理比较运算

的默认行为 TypeDecorator 是将任何表达式的“右手边”强制为同一类型。对于类似JSON的类型，这意味着所使用的任何运算符都必须对JSON有意义。在某些情况下，用户可能希望该类型在某些情况下的行为类似于JSON，而在其他情况下则是纯文本。一个例子是，如果想要为JSON类型处理like操作符。对于JSON结构，like没有意义，但是对于底层的文本表示它确实有意义。为了达到这个目的 JSONEncodedDict 我们需要 胁迫 使用将列转换为文本形式 cast() 或 type_coerce() 在尝试使用此运算符之前：

from sqlalchemy import type_coerce, String

stmt = select(my_table).where(
    type_coerce(my_table.c.json_data, String).like('%foo%'))

TypeDecorator 提供一个内置的系统，用于基于运算符处理类似的类型转换。如果我们希望经常在JSON对象被解释为字符串时使用LIKE运算符，我们可以通过重写 TypeDecorator.coerce_compared_value() 方法：

from sqlalchemy.sql import operators
from sqlalchemy import String

class JSONEncodedDict(TypeDecorator):

    impl = VARCHAR

    cache_ok = True

    def coerce_compared_value(self, op, value):
        if op in (operators.like_op, operators.not_like_op):
            return String()
        else:
            return self

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = json.dumps(value)

        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = json.loads(value)
        return value

上面只是处理类似“like”的操作符的一种方法。其他申请可能希望提高 NotImplementedError 对于对JSON对象（如“like”）没有意义的运算符，而不是自动强制转换为文本。

应用SQL级绑定/结果处理

如本节所示 扩充现有类型 ，sqlAlchemy允许在参数发送到语句时以及从数据库加载结果行时调用python函数，以便在值发送到数据库或从数据库加载时对其应用转换。也可以定义SQL级别的转换。这里的基本原理是，只有关系数据库包含强制应用程序和持久性格式之间的传入和传出数据所必需的一系列特定函数。示例包括使用数据库定义的加密/解密函数，以及处理地理数据的存储过程。PostGIS对PostgreSQL的扩展包括大量的SQL函数，这些函数是将数据强制转换为特定格式所必需的。

任何 TypeEngine ， UserDefinedType 或 TypeDecorator 子类可以包括 TypeEngine.bind_expression() 和/或 TypeEngine.column_expression() ，当定义为返回非“none”值时，它应返回 ColumnElement 要注入到SQL语句中的表达式，可以是环绕绑定参数，也可以是列表达式。例如，构建 Geometry 将应用PostGIS功能的类型 ST_GeomFromText 所有输出值和函数 ST_AsText 对于所有传入的数据，我们可以创建自己的子类 UserDefinedType 这些方法与 func ：：

from sqlalchemy import func
from sqlalchemy.types import UserDefinedType

class Geometry(UserDefinedType):
    def get_col_spec(self):
        return "GEOMETRY"

    def bind_expression(self, bindvalue):
        return func.ST_GeomFromText(bindvalue, type_=self)

    def column_expression(self, col):
        return func.ST_AsText(col, type_=self)

我们可以申请 Geometry 键入 Table 元数据并将其用于 select() 结构：

geometry = Table('geometry', metadata,
              Column('geom_id', Integer, primary_key=True),
              Column('geom_data', Geometry)
            )

print(select(geometry).where(
  geometry.c.geom_data == 'LINESTRING(189412 252431,189631 259122)'))

生成的SQL根据需要嵌入这两个函数。 ST_AsText 应用于columns子句，以便在传递到结果集之前通过函数运行返回值，以及 ST_GeomFromText 对绑定参数运行，以便转换传入的值：：

SELECT geometry.geom_id, ST_AsText(geometry.geom_data) AS geom_data_1
FROM geometry
WHERE geometry.geom_data = ST_GeomFromText(:geom_data_2)

这个 TypeEngine.column_expression() 方法与编译器的机制交互，以便SQL表达式不会干扰包装表达式的标记。例如，如果我们 select() 反对 label() 在表达式中，字符串标签被移动到包装表达式的外部：

print(select(geometry.c.geom_data.label('my_data')))

输出：

SELECT ST_AsText(geometry.geom_data) AS my_data
FROM geometry

另一个例子是我们装饰 BYTEA 提供一个 PGPString ，这将利用PostgreSQL pgcrypto 透明加密/解密值的扩展名：

from sqlalchemy import create_engine, String, select, func, \
        MetaData, Table, Column, type_coerce, TypeDecorator

from sqlalchemy.dialects.postgresql import BYTEA

class PGPString(TypeDecorator):
    impl = BYTEA

    cache_ok = True

    def __init__(self, passphrase):
        super(PGPString, self).__init__()

        self.passphrase = passphrase

    def bind_expression(self, bindvalue):
        # convert the bind's type from PGPString to
        # String, so that it's passed to psycopg2 as is without
        # a dbapi.Binary wrapper
        bindvalue = type_coerce(bindvalue, String)
        return func.pgp_sym_encrypt(bindvalue, self.passphrase)

    def column_expression(self, col):
        return func.pgp_sym_decrypt(col, self.passphrase)

metadata_obj = MetaData()
message = Table('message', metadata_obj,
                Column('username', String(50)),
                Column('message',
                    PGPString("this is my passphrase")),
            )

engine = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
with engine.begin() as conn:
    metadata_obj.create_all(conn)

    conn.execute(message.insert(), username="some user",
                                message="this is my message")

    print(conn.scalar(
            select(message.c.message).\
                where(message.c.username == "some user")
        ))

这个 pgp_sym_encrypt 和 pgp_sym_decrypt 函数应用于insert和select语句：

INSERT INTO message (username, message)
  VALUES (%(username)s, pgp_sym_encrypt(%(message)s, %(pgp_sym_encrypt_1)s))
  {'username': 'some user', 'message': 'this is my message',
    'pgp_sym_encrypt_1': 'this is my passphrase'}

SELECT pgp_sym_decrypt(message.message, %(pgp_sym_decrypt_1)s) AS message_1
  FROM message
  WHERE message.username = %(username_1)s
  {'pgp_sym_decrypt_1': 'this is my passphrase', 'username_1': 'some user'}

参见

PostGIS集成

重新定义和创建新的运算符

SQLAlchemy Core defines a fixed set of expression operators available to all column expressions. Some of these operations have the effect of overloading Python's built-in operators; examples of such operators include ColumnOperators.__eq__() (table.c.somecolumn == 'foo'), ColumnOperators.__invert__() (~table.c.flag), and ColumnOperators.__add__() (table.c.x + table.c.y). Other operators are exposed as explicit methods on column expressions, such as ColumnOperators.in_() (table.c.value.in_(['x', 'y'])) and ColumnOperators.like() (table.c.value.like('%ed%')).

核心表达式构造在所有情况下都会参考表达式的类型，以便确定现有运算符的行为，以及定位不属于内置集的其他运算符。这个 TypeEngine 基类定义根“比较”实现 Comparator 和许多特定类型提供了自己的此类子实现。用户定义的 Comparator 实现可以直接构建到特定类型的简单子类中，以便重写或定义新的操作。下面，我们创建一个 Integer 重写的子类 ColumnOperators.__add__() 操作员：

from sqlalchemy import Integer

class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def __add__(self, other):
            return self.op("goofy")(other)

上面的配置创建了一个新类 MyInt ，从而确定 TypeEngine.comparator_factory 属性引用一个新类，将 Comparator 与关联的类 Integer 类型。

用法：

>>> sometable = Table("sometable", metadata, Column("data", MyInt))
>>> print(sometable.c.data + 5)
sometable.data goofy :data_1

实施 ColumnOperators.__add__() 由所属的SQL表达式通过实例化 Comparator 把自己当作 expr 属性。表达式系统的机制是这样的：操作以递归方式继续，直到表达式对象生成新的SQL表达式构造。上面，我们也可以这么说 self.expr.op("goofy")(other) 而不是 self.op("goofy")(other) .

使用时 Operators.op() 对于返回布尔结果的比较操作， Operators.op.is_comparison 标志应设置为 True ：：

class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def is_frobnozzled(self, other):
            return self.op("--is_frobnozzled->", is_comparison=True)(other)

添加到 Comparator 在拥有的SQL表达式上使用 __getattr__ 方案，它公开添加到 Comparator 拥有 ColumnElement . 例如，要添加 log() 整型函数：

from sqlalchemy import Integer, func

class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def log(self, other):
            return func.log(self.expr, other)

使用上述类型：

>>> print(sometable.c.data.log(5))
log(:log_1, :log_2)

一元运算也是可能的。例如，为了添加PostgreSQL阶乘运算符的实现，我们将 UnaryExpression 与 custom_op 要生成阶乘表达式：

from sqlalchemy import Integer
from sqlalchemy.sql.expression import UnaryExpression
from sqlalchemy.sql import operators

class MyInteger(Integer):
    class comparator_factory(Integer.Comparator):
        def factorial(self):
            return UnaryExpression(self.expr,
                        modifier=operators.custom_op("!"),
                        type_=MyInteger)

使用上述类型：

>>> from sqlalchemy.sql import column
>>> print(column('x', MyInteger).factorial())
x !

参见

Operators.op()

TypeEngine.comparator_factory

创建新类型

这个 UserDefinedType 类作为一个简单的基类提供，用于定义全新的数据库类型。使用此项表示SQLAlchemy不知道的本机数据库类型。如果只需要python转换行为，请使用 TypeDecorator 相反。

Object Name	Description
UserDefinedType	用户定义类型的基。

class sqlalchemy.types.UserDefinedType

需要注意的是，修改后的数据库类型具有额外的Python行为，包括基于 TypeDecorator 以及其他用户定义的数据类型子类，在数据库模式中没有任何表示。当使用数据库时，在 反映数据库对象 ，SQLAlchemy使用了一个固定的映射，该映射将数据库服务器报告的数据类型信息链接到SQLAlchemy数据类型对象。例如，如果我们在PostgreSQL模式内部查看特定数据库列的定义，我们可能会收到返回的字符串 "VARCHAR" . SQLAlchemy的PostgreSQL方言有一个链接字符串名称的硬编码映射 "VARCHAR" 对SQLAlchemy VARCHAR 类，当我们发出一个 Table('my_table', m, autoload_with=engine) , the Column 对象中的 VARCHAR 呈现在里面。

这意味着如果 Table 如果我们创建一个新的 Table 反对新的 MetaData 在其他地方使用反射的此数据库表的集合，它将不具有此数据类型。例如：：

>>> from sqlalchemy import Table, Column, MetaData, create_engine, PickleType, Integer
>>> metadata = MetaData()
>>> my_table = Table("my_table", metadata, Column('id', Integer), Column("data", PickleType))
>>> engine = create_engine("sqlite://", echo='debug')
>>> my_table.create(engine)
INFO sqlalchemy.engine.base.Engine
CREATE TABLE my_table (
    id INTEGER,
    data BLOB
)

上面，我们利用 PickleType ，这是一个 TypeDecorator 它在 LargeBinary 数据类型，它在SQLite上对应于数据库类型 BLOB . 在CREATE表中，我们看到 BLOB 使用了数据类型。SQLite数据库对 PickleType 我们用过。

如果我们看看 my_table.c.data.type ，因为这是我们直接创建的Python对象，因此 PickleType ：：

>>> my_table.c.data.type
PickleType()

但是，如果我们创建另一个 Table 使用反射，使用 PickleType 在我们创建的SQLite数据库中没有表示；而是返回 BLOB ：：

>>> metadata_two = MetaData()
>>> my_reflected_table = Table("my_table", metadata_two, autoload_with=engine)
INFO sqlalchemy.engine.base.Engine PRAGMA main.table_info("my_table")
INFO sqlalchemy.engine.base.Engine ()
DEBUG sqlalchemy.engine.base.Engine Col ('cid', 'name', 'type', 'notnull', 'dflt_value', 'pk')
DEBUG sqlalchemy.engine.base.Engine Row (0, 'id', 'INTEGER', 0, None, 0)
DEBUG sqlalchemy.engine.base.Engine Row (1, 'data', 'BLOB', 0, None, 0)

>>> my_reflected_table.c.data.type
BLOB()

通常，当应用程序定义显式 Table 对于自定义类型的元数据，不需要使用表反射，因为 Table 元数据已存在。但是，对于应用程序或它们的组合需要同时使用显式和 Table 元数据，包括自定义的、Python级别的数据类型，以及 Table 设置其 Column 对象从数据库中反映出来，但是仍然需要显示自定义数据类型的附加Python行为，必须采取额外的步骤来实现这一点。

最简单的方法是重写特定列，如中所述 覆盖反射列 . 在这种技术中，我们简单地将反射与显式结合使用 Column 要对其使用自定义或修饰数据类型的列的对象：

>>> metadata_three = MetaData()
>>> my_reflected_table = Table("my_table", metadata_three, Column("data", PickleType), autoload_with=engine)

这个 my_reflected_table 对象将被反射，并将从SQLite数据库加载“id”列的定义。但是对于“data”列，我们用显式 Column 包含所需的Python数据类型的定义 PickleType . 反射过程会留下这个 Column 对象完好：

>>> my_reflected_table.c.data.type
PickleType()

从数据库本机类型对象转换为自定义数据类型的一种更精细的方法是使用 DDLEvents.column_reflect() 事件处理程序。如果我们知道我们想要一切 BLOB 数据类型实际上是 PickleType ，我们可以制定一个全面的规则：

from sqlalchemy import BLOB
from sqlalchemy import event
from sqlalchemy import PickleType
from sqlalchemy import Table

@event.listens_for(Table, "column_reflect")
def _setup_pickletype(inspector, table, column_info):
    if isinstance(column_info["type"], BLOB):
        column_info["type"] = PickleType()

当调用上述代码时 之前 发生任何表反射（注意应该调用它 只有一次 在应用程序中，因为这是一个全局规则） Table 包含一个列 BLOB 数据类型，则生成的数据类型将存储在 Column 对象AS PickleType .

在实践中，上述基于事件的方法可能会有额外的规则，以便只影响那些数据类型重要的列，例如表名和列名的查找表，或者其他启发式方法，以便准确地确定哪些列应该使用In Python数据类型建立。