重写类型编译¶

一个常见的需求是强制更改类型的“字符串”版本，即在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 需要与数据库进行数据交换。

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

允许创建向现有类型添加附加功能的类型。

此方法优先于直接对SQLAlchemy的内置类型进行子类化，因为它可以确保底层类型的所有必需功能都保持在适当的位置。

典型用法：

import sqlalchemy.types as types

class MyType(types.TypeDecorator):
    '''Prefixes Unicode values with "PREFIX:" on the way in and
    strips it off on the way out.
    '''

    impl = types.Unicode

    cache_ok = True

    def process_bind_param(self, value, dialect):
        return "PREFIX:" + value

    def process_result_value(self, value, dialect):
        return value[7:]

    def copy(self, **kw):
        return MyType(self.impl.length)

班级级别 impl 属性是必需的，并且可以引用任何 TypeEngine 班级。或者 load_dialect_impl() 方法可用于根据给定的方言提供不同的类型类；在本例中 impl 变量可以引用 TypeEngine 作为占位符。

这个 TypeDecorator.cache_ok 类级标志指示此自定义 TypeDecorator 作为缓存键的一部分使用是安全的。此标志默认为 None 当SQL编译器尝试为使用此类型的语句生成缓存键时，最初将生成警告。如果 TypeDecorator 不能保证每次都会生成相同的绑定/结果行为和SQL生成，此标志应设置为 False ；否则，如果该类每次都产生相同的行为，则可能会将其设置为 True 。看见 TypeDecorator.cache_ok 有关此功能如何工作的进一步说明，请参阅。

接收与使用的最终类型不同的python类型的类型可能希望定义 TypeDecorator.coerce_compared_value() 方法。这用于在将Python对象强制为表达式中的绑定参数时向表达式系统提供提示。考虑以下表达式：

mytable.c.somecol + datetime.date(2009, 5, 15)

如果“somecol”是 Integer 变量，我们在做日期算术是有意义的，上面的数据通常被数据库解释为给给定的日期增加了几天。表达式系统通过不尝试将“date（）”值强制为面向整数的绑定参数来做正确的事情。

但是，在 TypeDecorator ，我们通常将传入的python类型更改为新类型。- TypeDecorator 默认情况下，将“强制”非类型化的端与自身的类型相同。如下图所示，我们定义了一个“epoch”类型，该类型将日期值存储为整数：

class MyEpochType(types.TypeDecorator):
    impl = types.Integer

    epoch = datetime.date(1970, 1, 1)

    def process_bind_param(self, value, dialect):
        return (value - self.epoch).days

    def process_result_value(self, value, dialect):
        return self.epoch + timedelta(days=value)

我们的表达 somecol + date 如果使用上述类型，则右侧的“日期”也将被视为 MyEpochType .

此行为可以通过 TypeDecorator.coerce_compared_value() 方法，返回应用于表达式值的类型。下面我们将其设置为一个整数值将被视为 Integer ，任何其他值假定为日期，并将被视为 MyEpochType ：：

def coerce_compared_value(self, op, value):
    if isinstance(value, int):
        return Integer()
    else:
        return self

警告

请注意 behavior of coerce_compared_value is not inherited by default from that of the base type . 如果 TypeDecorator 正在为某些类型的运算符扩充需要特殊逻辑的类型，此方法 must 被重写。一个关键的例子是当装饰 JSON 和 JSONB 类型；默认规则 TypeEngine.coerce_compared_value() 应用于处理索引操作等运算符：

class MyJsonType(TypeDecorator):
    impl = postgresql.JSON

    cache_ok = True

    def coerce_compared_value(self, op, value):
        return self.impl.coerce_compared_value(op, value)

如果没有上述步骤，则索引操作，例如 mycol['foo'] 将导致索引值 'foo' 将被JSON编码。

类签名

class sqlalchemy.types.TypeDecorator (sqlalchemy.sql.expression.SchemaEventTarget, sqlalchemy.types.TypeEngine)

class Comparator(expr)¶

A Comparator 这是特定于 TypeDecorator .

用户定义的 TypeDecorator 类通常不需要修改它。

类签名

class sqlalchemy.types.TypeDecorator.Comparator (sqlalchemy.types.Comparator)

method sqlalchemy.types.TypeDecorator.Comparator.operate(op, *other, **kwargs)¶

对参数进行运算。

这是最低级别的操作，提升 NotImplementedError 默认情况下。

在子类上覆盖此项可以允许将公共行为应用于所有操作。例如，重写 ColumnOperators 申请 func.lower() 左右两侧：

class MyComparator(ColumnOperators):
    def operate(self, op, other):
        return op(func.lower(self), func.lower(other))

参数

• op¶ -- 操作员可调用。

• *other¶ -- 操作的“另一方”。对于大多数操作，将是单个标量。

• **kwargs¶ -- 修饰语。这些可由特殊操作员通过，如 ColumnOperators.contains() .

method sqlalchemy.types.TypeDecorator.Comparator.reverse_operate(op, other, **kwargs)¶

对参数进行反向运算。

用法与 operate() .

method sqlalchemy.types.TypeDecorator.__init__(*args, **kwargs)¶

构建一个 TypeDecorator .

此处发送的参数将传递给分配给 impl 类级属性，假定 impl 是可调用的，并且结果对象被分配给 self.impl 实例属性（从而覆盖相同名称的类属性）。

如果等级 impl 不是可调用的（异常情况），它将被分配给与“原样”相同的实例属性，忽略传递给构造函数的参数。

子类可以重写此项以自定义 self.impl 完全。

method sqlalchemy.types.TypeDecorator.bind_expression(bindparam)¶

给定一个绑定值（即 BindParameter 实例），返回一个SQL表达式。

这通常是一个在语句中包装现有绑定参数的SQL函数。它用于特殊数据类型，这些数据类型需要将文本包装在某些特殊的数据库函数中，以便将应用程序级值强制为特定于数据库的格式。它类似于SQL TypeEngine.bind_processor() 方法。

该方法在语句编译时进行评估，而不是在语句构造时进行评估。

请注意，此方法在实现时，应始终返回完全相同的结构，而不返回任何条件逻辑，因为它可以在针对任意数量的绑定参数集的executeMany（）调用中使用。

参见

应用SQL级绑定/结果处理

method sqlalchemy.types.TypeDecorator.bind_processor(dialect)¶

为给定的提供绑定值处理函数 Dialect .

这是实现 TypeEngine 约束值转换合同。 TypeDecorator 将包装用户定义的 process_bind_param() 在这里。

用户定义的代码可以直接重写这个方法，尽管它可能最适合使用 process_bind_param() 以便处理所提供的 self.impl 保持。

参数

dialect¶ -- 方言实例正在使用中。

此方法与 result_processor() 此类的方法。

attribute sqlalchemy.types.TypeDecorator.cache_ok = None¶

指示语句是否使用此 TypeDecorator 都是“可以安全缓存”的。

默认值 None 将发出警告，然后不允许缓存包含此类型的语句。设置为 False 在没有警告的情况下完全禁止缓存使用此类型的语句。当设置为时 True ，则对象的类和来自其状态的选定元素将用作缓存键的一部分，例如：：

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

    def __init__(self, choices):
        self.choices = tuple(choices)
        self.internal_only = True

上述类型的缓存键等效于：：

(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

缓存方案将从与 __init__() 方法。上面，“CHOICES”属性成为缓存键的一部分，但“INTERNAL_ONLY”没有，因为没有名为“INTERNAL_ONLY”的参数。

对可缓存元素的要求是它们是可哈希的，并且它们还指示每次为给定缓存值使用此类型的表达式呈现相同的SQL。

1.4.14 新版功能: -添加了 cache_ok 允许缓存的某些可配置性的标志 TypeDecorator 上课。

参见

SQL编译缓存

method sqlalchemy.types.TypeDecorator.coerce_compared_value(op, value)¶

为表达式中的“强制”python值建议类型。

默认情况下，返回self。当使用此类型的对象位于表达式的左侧或右侧，而普通的python对象尚未指定sqlacalchemy类型时，表达式系统将调用此方法：：

expr = table.c.somecolumn + 35

如果以上，如果 somecolumn 使用此类型，将使用值调用此方法 operator.add 和 35 . 返回值是应用于 35 对于这个特殊的操作。

attribute sqlalchemy.types.TypeDecorator.coerce_to_is_types = (<class 'NoneType'>,)¶

指定在表达式级别应强制为“is<constant>”的python类型，当使用 == （同样） IS NOT 与 != ）

对于大多数SQLAlchemy类型，这包括 NoneType 以及 bool .

TypeDecorator 将此列表修改为仅包含 NoneType ，因为处理布尔类型的typedecorator实现是常见的。

习俗 TypeDecorator 类可以重写此属性以返回空元组，在这种情况下，不会将任何值强制为常量。

method sqlalchemy.types.TypeDecorator.column_expression(column)¶

给定select列表达式，返回包装SQL表达式。

这通常是一个SQL函数，它在select语句的columns子句中呈现列表达式。它用于特殊数据类型，这些数据类型要求将列包装在某些特殊的数据库函数中，以便在发送回应用程序之前强制该值。它类似于SQL TypeEngine.result_processor() 方法。

该方法在语句编译时进行评估，而不是在语句构造时进行评估。

参见

应用SQL级绑定/结果处理

attribute sqlalchemy.types.TypeDecorator.comparator_factory¶

A Comparator 类的新实例，该类将应用于由拥有 ColumnElement 对象。

这个 comparator_factory 属性是核心表达式系统在执行列和SQL表达式操作时引用的钩子。当一个 Comparator 类与此属性关联，它允许自定义重新定义所有现有运算符，以及定义新运算符。现有运算符包括Python运算符重载提供的运算符，例如 ColumnOperators.__add__() 和 ColumnOperators.__eq__() ，这些属性作为的标准属性提供 ColumnOperators 比如 ColumnOperators.like() 和 ColumnOperators.in_() 。

通过对现有类型进行简单的子类化，或者通过使用 TypeDecorator 。请参阅文档部分 重新定义和创建新的运算符 举个例子。

method sqlalchemy.types.TypeDecorator.compare_values(x, y)¶

给定两个值，比较它们是否相等。

默认情况下，此调用 TypeEngine.compare_values() 在底层“impl”中，通常使用python equals运算符 == .

ORM使用此函数将原始加载值与截取的“已更改”值进行比较，以确定是否发生了净更改。

method sqlalchemy.types.TypeDecorator.copy(**kw)¶

生成此的副本 TypeDecorator 实例。

这是一个肤浅的副本，用于完成 TypeEngine 合同。它通常不需要重写，除非用户定义 TypeDecorator 具有应进行深度复制的本地状态。

method sqlalchemy.types.TypeDecorator.get_dbapi_type(dbapi)¶

返回由此表示的DBAPI类型对象 TypeDecorator .

默认情况下，此调用 TypeEngine.get_dbapi_type() 基本的“impl”。

method sqlalchemy.types.TypeDecorator.literal_processor(dialect)¶

为给定的提供文本处理函数 Dialect .

这里的子类通常会覆盖 TypeDecorator.process_literal_param() 而不是直接用这种方法。

默认情况下，此方法使用 TypeDecorator.process_bind_param() 如果实现了该方法，那么 TypeDecorator.process_literal_param() 不是。这里的基本原理是 TypeDecorator 通常处理数据库表示层之上的数据的python转换。值转换为 TypeDecorator.process_bind_param() 然后，基础类型将处理它是否需要作为绑定参数呈现给DBAPI，还是作为内联SQL值呈现给数据库。

0.9.0 新版功能.

method sqlalchemy.types.TypeDecorator.load_dialect_impl(dialect)¶

返回A TypeEngine 与方言相对应的对象。

这是一个终端用户覆盖钩子，可以根据给定的方言提供不同的类型。它被 TypeDecorator 执行 type_engine() 以帮助确定最终应为给定的 TypeDecorator .

默认情况下返回 self.impl .

method sqlalchemy.types.TypeDecorator.process_bind_param(value, dialect)¶

接收要转换的绑定参数值。

子类重写此方法以返回应传递给基础的值 TypeEngine 对象，从那里到DBAPI execute() 方法。

该操作可以是执行自定义行为所需的任何操作，例如转换或序列化数据。这也可以用作验证逻辑的钩子。

此操作的设计应考虑到反向操作，这将是此类的过程结果值方法。

参数

• value¶ -- 要操作的数据，属于子类中此方法所期望的任何类型。可以是 None .

• dialect¶ -- 这个 Dialect 在使用中。

method sqlalchemy.types.TypeDecorator.process_literal_param(value, dialect)¶

接收要在语句内内联呈现的文本参数值。

当编译器在不使用绑定的情况下呈现文本值时，通常在DDL中使用此方法，例如在列的“服务器默认值”或检查约束中的表达式中。

返回的字符串将呈现到输出字符串中。

0.9.0 新版功能.

method sqlalchemy.types.TypeDecorator.process_result_value(value, dialect)¶

接收要转换的结果行列值。

子类应该实现这个方法来操作从数据库中获取的数据。

子类重写此方法以返回应传递回应用程序的值，给定的值已由基础处理 TypeEngine 对象，最初来自dbapi cursor方法 fetchone() 或者类似的。

该操作可以是执行自定义行为所需的任何操作，例如转换或序列化数据。这也可以用作验证逻辑的钩子。

参数

• value¶ -- 要操作的数据，属于子类中此方法所期望的任何类型。可以是 None .

• dialect¶ -- 这个 Dialect 在使用中。

此操作应设计为通过此类的“process-bind-param”方法可逆。

method sqlalchemy.types.TypeDecorator.result_processor(dialect, coltype)¶

为给定的提供结果值处理函数 Dialect .

这是实现 TypeEngine 结果值转换合同。 TypeDecorator 将包装用户定义的 process_result_value() 在这里。

用户定义的代码可以直接重写这个方法，尽管它可能最适合使用 process_result_value() 以便处理所提供的 self.impl 保持。

参数

• dialect¶ -- 方言实例正在使用中。

• coltype¶ -- SQLAlchemy数据类型

此方法与 bind_processor() 此类的方法。

method sqlalchemy.types.TypeDecorator.type_engine(dialect)¶

返回特定方言 TypeEngine 这个的实例 TypeDecorator .

在大多数情况下，这将返回方言适配的 TypeEngine 表示的类型 self.impl 。利用 dialect_impl() 。通过重写以下内容，可以在此处自定义行为 load_dialect_impl() 。

typedecorator配方¶

几把钥匙 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

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

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

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

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

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

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

json_type = MutableDict.as_mutable(JSONEncodedDict)

class MyClass(Base):
    #  ...

    json_data = Column(json_type)

from sqlalchemy import type_coerce, String

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

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

应用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'}

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

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 !

创建新类型¶

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

class sqlalchemy.types.UserDefinedType¶

用户定义类型的基。

这应该是新类型的基础。注意，在大多数情况下， TypeDecorator 可能更合适：

import sqlalchemy.types as types

class MyType(types.UserDefinedType):
    def __init__(self, precision = 8):
        self.precision = precision

    def get_col_spec(self, **kw):
        return "MYTYPE(%s)" % self.precision

    def bind_processor(self, dialect):
        def process(value):
            return value
        return process

    def result_processor(self, dialect, coltype):
        def process(value):
            return value
        return process

一旦生成类型，它立即可用：

table = Table('foo', meta,
    Column('id', Integer, primary_key=True),
    Column('data', MyType(16))
    )

这个 get_col_spec() 在大多数情况下，方法将接收关键字参数 type_expression 它引用正在编译的类型的所属表达式，例如 Column 或 cast() 构建。仅当方法接受关键字参数（例如 **kw ）在它的参数签名中，内省用于检查这一点，以便支持此函数的遗留形式。

1.0.0 新版功能: 所属表达式传递给 get_col_spec() 方法通过关键字参数 type_expression ，如果它收到 **kw 在它的签名上。

类签名

class sqlalchemy.types.UserDefinedType (sqlalchemy.types.TypeEngine)

method sqlalchemy.types.UserDefinedType.coerce_compared_value(op, value)¶

为表达式中的“强制”python值建议类型。

的默认行为 UserDefinedType 与…相同 TypeDecorator ；默认返回 self ，假定比较值应强制为与此值相同的类型。见 TypeDecorator.coerce_compared_value() 更多细节。

使用自定义类型和反射¶

需要注意的是，修改后的数据库类型具有额外的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数据类型建立。