自定义类型¶

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

覆盖类型编译¶

一个常见的需求是强制更改类型的“字符串”版本，即在 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，它将针对除 SQLite 之外的所有后端生成字符串 BINARY，在这种情况下，它将生成 BLOB。

请参阅更改类型的编译一节，该小节是自定义 SQL 构造和编译扩展，以获取更多示例。

扩充现有类型¶

TypeDecorator 允许创建自定义类型，这些类型将 bind-parameter 和 result-processing 行为添加到现有类型对象中。当需要对数据库进行额外的 Python 内数据封送和/或从数据库进行数据编组时，使用它。

注意

TypeDecorator 的 bind 和 result-processing 是托管类型已执行的处理的补充，托管类型由 SQLAlchemy 基于每个 DBAPI 进行自定义，以执行特定于该 DBAPI 的处理。虽然可以通过直接子类化替换给定类型的这种处理，但在实践中从来不需要，并且 SQLAlchemy 不再支持将其作为公共用例。

对象名称	描述
类型装饰器	允许创建向现有类型添加附加功能的类型。

类 sqlalchemy.types 中。类型装饰器¶

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

此方法比 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（） method 可用于根据 dialect 提供不同的类型类鉴于;在这种情况下，impl 变量可以引用 TypeEngine 作为占位符。

TypeDecorator.cache_ok类级标志指示此自定义 TypeDecorator 是否可以安全地用作缓存键的一部分。此标志默认为 None，当 SQL 编译器尝试为使用此类型的语句生成缓存键时，它最初将生成警告。如果不能保证 TypeDecorator 每次都产生相同的 bind/result 行为和 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

    cache_ok = True

    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 的上述类型将强制右侧的 “date” 也被视为 MyEpochType。

此行为可以通过 TypeDecorator.coerce_compared_value() 方法，该方法返回应用于表达式值的 type。下面我们对其进行设置，以便将 Integer 值视为 Integer，而任何其他值都假定为日期并被视为 MyEpochType：

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

警告

请注意，coerce_compared_value 的行为不是继承的默认情况下，从 base 类型的 base 类型开始。如果 TypeDecorator 正在增强需要某些类型运算符的特殊逻辑的类型，则必须覆盖此方法。一个关键示例是装饰 JSON 和 JSONB 类型时; TypeEngine.coerce_compared_value() 应该使用的默认规则来处理 index作等运算符：

from sqlalchemy import JSON
from sqlalchemy import TypeDecorator


class MyJsonType(TypeDecorator):
    impl = JSON

    cache_ok = True

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

如果没有上述步骤，诸如 mycol['foo'] 之类的索引作将导致索引值 'foo' 进行 JSON 编码。

同样，当使用 ARRAY 数据类型时，索引作的类型强制（例如 mycol[5]）也由 TypeDecorator.coerce_compared_value() 处理，其中同样，除非特定运算符需要特殊规则，否则简单的覆盖就足够了：

from sqlalchemy import ARRAY
from sqlalchemy import TypeDecorator


class MyArrayType(TypeDecorator):
    impl = ARRAY

    cache_ok = True

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

类签名

类 sqlalchemy.types.TypeDecorator （ sqlalchemy.sql.expression.SchemaEventTarget ， sqlalchemy.types.ExternalType， sqlalchemy.types.TypeEngine）

属性 sqlalchemy.types.TypeDecorator. cache_ok：boolNone = None¶

继承自ExternalType 的 ExternalType.cache_ok属性

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

默认值 None 将发出警告，然后不允许缓存包含此类型的语句。设置为 False 可完全禁止缓存使用此类型的语句，而不会显示警告。当设置为 True 时，对象的类和从其 state 将用作缓存键的一部分。例如，使用 TypeDecorator 中：

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

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

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

>>> MyType(["a", "b", "c"])._static_cache_key
(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

缓存方案将从类型中提取与 __init__（） 方法中的参数名称相对应的属性。在上面，“choices”属性成为缓存键的一部分，但“internal_only”不是，因为没有名为“internal_only”的参数。

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

为了适应引用不可哈希结构（如字典、集合和列表）的数据类型，可以通过将可哈希结构分配给名称与参数名称对应的属性来使这些对象“可缓存”。例如，接受查找值字典的数据类型可能会将其发布为一系列排序的 Tuples。给定一个以前不可缓存的类型为：

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    this is the non-cacheable version, as "self.lookup" is not
    hashable.

    """

    def __init__(self, lookup):
        self.lookup = lookup

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self.lookup" ...

其中 “lookup” 是字典。该类型将无法生成缓存键：

>>> type_ = LookupType({"a": 10, "b": 20})
>>> type_._static_cache_key
<stdin>:1: SAWarning: UserDefinedType LookupType({'a': 10, 'b': 20}) will not
produce a cache key because the ``cache_ok`` flag is not set to True.
Set this flag to True if this type object's state is safe to use
in a cache key, or False to disable this warning.
symbol('no_cache')

如果我们确实设置了这样的缓存键，它将不可用。我们将得到一个包含字典的元组结构，它本身不能用作 “缓存字典” 中的键，例如 SQLAlchemy 的语句缓存，因为 Python 字典是不可哈希的：

>>> # set cache_ok = True
>>> type_.cache_ok = True

>>> # this is the cache key it would generate
>>> key = type_._static_cache_key
>>> key
(<class '__main__.LookupType'>, ('lookup', {'a': 10, 'b': 20}))

>>> # however this key is not hashable, will fail when used with
>>> # SQLAlchemy statement cache
>>> some_cache = {key: "some sql value"}
Traceback (most recent call last): File "<stdin>", line 1,
in <module> TypeError: unhashable type: 'dict'

通过将排序的 Tuples 元组分配给 “.lookup” 属性，可以使类型可缓存：

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    The dictionary is stored both as itself in a private variable,
    and published in a public variable as a sorted tuple of tuples,
    which is hashable and will also return the same value for any
    two equivalent dictionaries.  Note it assumes the keys and
    values of the dictionary are themselves hashable.

    """

    cache_ok = True

    def __init__(self, lookup):
        self._lookup = lookup

        # assume keys/values of "lookup" are hashable; otherwise
        # they would also need to be converted in some way here
        self.lookup = tuple((key, lookup[key]) for key in sorted(lookup))

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self._lookup" ...

在上面，的 LookupType({"a": 10, "b": 20}) 缓存键将是：

>>> LookupType({"a": 10, "b": 20})._static_cache_key
(<class '__main__.LookupType'>, ('lookup', (('a', 10), ('b', 20))))

在 1.4.14 版本加入： - 添加了 cache_ok 标志以允许对 TypeDecorator 类进行一些可配置性缓存。

1.4.28 版本的新Function： - 添加了 ExternalType mixin，它将 cache_ok 标志推广到 TypeDecorator 和 UserDefinedType 类。

另请参阅

SQL 编译缓存

类 Comparator¶

特定于 TypeDecorator 的

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

类签名

类 sqlalchemy.types.TypeDecorator.Comparator （sqlalchemy.types.Comparator）

method sqlalchemy.types.TypeDecorator.Comparator. operate（op： OperatorType， *other： Any， **kwargs： Any→ ColumnElement （列元素）[_CT]¶

对参数进行作。

这是最低级别的作，加注 NotImplementedError 的 MethodS 错误。

在子类上覆盖 this 可以允许将通用行为应用于所有作。例如，重写 ColumnOperators 要将 func.lower（） 应用于左侧和右侧：

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

参数

op¶- 运算符可调用。
other¶ ——作的 'other' 端。对于大多数作，将是单个标量。
**kwargs¶ —— 修饰符。这些可以通过特殊运算符（如 ColumnOperators.contains（）））传递。

方法 sqlalchemy.types.TypeDecorator.Comparator. reverse_operate（op： OperatorType， other： Any， **kwargs： Any→ ColumnElement[_CT]¶

Reverse 对参数进行作。

用法与 operate（） 相同。

方法 sqlalchemy.types.TypeDecorator. __init__（*args： Any， **kwargs： Any）¶

构造一个 TypeDecorator。

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

如果类级别 impl 不是可调用的（不常见的情况），它将被 'as-is' 分配给相同的实例属性，忽略传递给构造函数的那些参数。

子类可以覆盖 this 以完全自定义 self.impl 的生成。

方法 sqlalchemy.types.TypeDecorator. bind_expression（bindparam： BindParameter[_T]）→ColumnElement[_T]无¶

给定一个 bind 值（即 BindParameter 实例），返回一个 SQL 表达式，该表达式通常会包装给定的参数。

注意

在语句的 SQL 编译阶段，在呈现 SQL 字符串时调用此方法。不一定 针对特定值调用，并且不应与 TypeDecorator.process_bind_param() 方法，这是在语句执行时处理传递给特定参数的实际值的更典型的方法。

TypeDecorator 的子类可以覆盖此方法，以便为类型提供自定义绑定表达式行为。此 implementation 将替换底层 implementation 类型的 implementation。

方法 sqlalchemy.types.TypeDecorator. bind_processor（dialect： Dialect）→_BindProcessorType[_T]无¶

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

这是实现 TypeEngine 的方法合约进行绑定值转换，通常通过 TypeEngine.bind_processor（） 方法。

注意

用户定义的 TypeDecorator 子类应该不实现此方法，而应该实现 TypeDecorator.process_bind_param() 以便维护 implementation 类型提供的 “inner” 处理。

参数: dialect¶– 正在使用的 Dialect 实例。

method sqlalchemy.types.TypeDecorator. coerce_compared_value（op：OperatorTypeNone， value： Any）→ Any¶

为表达式中的 'coerced' Python 值建议类型。

默认情况下，返回 self。当使用此类型的对象位于表达式的左侧或右侧时，表达式系统将调用此方法，该对象尚未分配 SQLAlchemy 类型：

expr = table.c.somecolumn + 35

在上面，如果 somecolumn 使用此类型，则将使用值 operator.add 调用此方法和 35.返回值是此特定作的 35 应该使用的 SQLAlchemy 类型。

属性 sqlalchemy.types.TypeDecorator. coerce_to_is_types： Sequence[Type[Any]] = （<class 'NoneType'>，）¶

使用 == 进行比较时，指定应在表达式级别强制为 “IS <constant>” 的 Python 类型（对于 IS NOT 与 ！= 结合使用）。

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

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

自定义 TypeDecorator 类可以覆盖此属性以返回空 Tuples，在这种情况下，不会将任何值强制转换为常量。

方法 sqlalchemy.types.TypeDecorator. column_expression（column： ColumnElement[_T]）→ColumnElement[_T]无¶

给定一个 SELECT 列表达式，返回一个包装 SQL 表达式。

注意

在语句的 SQL 编译阶段，在呈现 SQL 字符串时调用此方法。它不叫针对特定值，并且不应与 TypeDecorator.process_result_value() 方法，这是在语句执行时间之后处理结果行中返回的实际值的更典型的方法。

TypeDecorator 的子类可以覆盖此方法，以便为类型提供自定义列表达式行为。此 implementation 将替换底层 implementation 类型的 implementation。

TypeEngine.column_expression() 请参阅以获取该方法用法的完整描述。

属性 sqlalchemy.types.TypeDecorator. comparator_factory：_ComparatorFactory[Any]¶

一个 Comparator 类，该类将应用于通过拥有 ColumnElement 执行的作对象。

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

通过对现有类型进行简单的子类化，或者通过使用 TypeDecorator 来允许对此钩子的基本使用。有关示例，请参阅文档部分重新定义和创建新运算符。

方法 sqlalchemy.types.TypeDecorator. compare_values（x： Any， y： Any）→ bool¶

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

默认情况下，这会调用 TypeEngine.compare_values（） 底层的 “impl” 中，而使用 Python 等于运算符 ==。

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

method sqlalchemy.types.TypeDecorator. copy（**kw： Any）→ Self¶

生成此 TypeDecorator 实例的副本。

这是一个浅层副本，用于履行 TypeEngine 合同的一部分。它通常不需要被覆盖，除非用户定义的 TypeDecorator 具有应深层复制的本地状态。

方法 sqlalchemy.types.TypeDecorator. get_dbapi_type（dbapi： module）→AnyNone¶

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

默认情况下，这调用底层 “impl” 的 TypeEngine.get_dbapi_type（） 。

方法 sqlalchemy.types.TypeDecorator. literal_processor（dialect： Dialect）→_LiteralProcessorType[_T]无¶

为给定的 方言。

这是实现 TypeEngine 的方法字面值转换的协定，通常通过方法 TypeEngine.literal_processor() 。

注意

用户定义的 TypeDecorator 子类应该不实现此方法，而应该实现 TypeDecorator.process_literal_param() 以便维护 implementation 类型提供的 “inner” 处理。

方法 sqlalchemy.types.TypeDecorator. load_dialect_impl（dialect： Dialect）→ TypeEngine[Any]¶

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

这是一个最终用户覆盖钩子，可用于根据给定的方言提供不同的类型。它由 type_engine（） 的 TypeDecorator 实现使用帮助确定最终应返回的类型对于给定的 TypeDecorator 。

默认情况下，返回 self.impl。

方法 sqlalchemy.types.TypeDecorator. process_bind_param（value：_TNone， dialect： Dialect）→ Any¶

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

TypeDecorator 的自定义子类应该覆盖此方法，以便为传入的数据值提供自定义行为。此方法在语句执行时调用，并传递文本 Python 数据值，该值将与语句中的绑定参数相关联。

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

参数

value¶- 要作的数据，子类中此方法所需的任何类型的数据。可以是 None。
dialect¶—— 正在使用的 Dialect。

另请参阅

扩充现有类型

TypeDecorator.process_result_value()

方法 sqlalchemy.types.TypeDecorator. process_literal_param（value：_TNone， dialect： Dialect）→ str¶

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

注意

此方法在 SQL 编译阶段调用语句。与其他 SQL 不同编译方法，则会传递一个特定的 Python 值呈现为 String 的字符串。但是，它不应与 TypeDecorator.process_bind_param() 方法，这是在语句执行时处理传递给特定参数的实际值的更典型的方法。

TypeDecorator 的自定义子类应该覆盖此方法，以便为在渲染为 Literals 的特殊情况下的传入数据值提供自定义行为。

返回的字符串将渲染到输出字符串中。

方法 sqlalchemy.types.TypeDecorator. process_result_value（value：AnyNone， dialect： Dialect）→_TNone¶

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

TypeDecorator 的自定义子类应该覆盖此方法，以便为来自数据库的结果行中接收的数据值提供自定义行为。此方法在结果提取时调用，并传递从数据库结果行中提取的文本 Python 数据值。

该作可以是执行自定义行为所需的任何作，例如转换或反序列化数据。

参数

value¶- 要作的数据，子类中此方法所需的任何类型的数据。可以是 None。
dialect¶—— 正在使用的 Dialect。

另请参阅

扩充现有类型

TypeDecorator.process_bind_param()

方法 sqlalchemy.types.TypeDecorator. result_processor（dialect：方言， 同类型： Any）→_ResultProcessorType[_T]无¶

为给定的 方言。

这是实现 TypeEngine 的方法合约进行绑定值转换，通常通过 TypeEngine.result_processor（） 方法。

注意

用户定义的 TypeDecorator 子类应该不实现此方法，而应该实现 TypeDecorator.process_result_value() 以便维护 implementation 类型提供的 “inner” 处理。

参数

dialect¶– 正在使用的 Dialect 实例。
coltype¶ – SQLAlchemy 数据类型

attribute sqlalchemy.types.TypeDecorator. sort_key_function：Callable[[Any]，Any]无¶

一个排序函数，可以作为键传递给 sorted。

默认值 None 表示此类型存储的值是自排序的。

在 1.3.8 版本加入.

方法 sqlalchemy.types.TypeDecorator. type_engine（dialect： Dialect）→ TypeEngine[Any]¶

返回此 TypeDecorator 的特定于方言的 TypeEngine 实例。

在大多数情况下，这将返回由 self.impl 表示的 TypeEngine 类型的方言适应形式。使用 dialect_impl（）。 可以通过重写 load_dialect_impl（） 中。

TypeDecorator 配方¶

下面是一些关键的 TypeDecorator 配方。

将编码的字符串强制转换为 Unicode¶

关于 Unicode 类型的一个常见混淆来源是它仅用于处理 Python 端的 Python unicode 对象，这意味着如果使用 Python 2 而不是 3，则作为绑定参数传递给它的值必须采用 u'some string' 的形式。它执行的编码 / 解码功能只是为了满足正在使用的 DBAPI 的需要，并且主要是一个私有的实现细节。

可以安全地接收 Python 字节字符串的类型用例，即包含非 ASCII 字符且不是 u'' 的字符串对象，可以使用 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

四舍五入数字¶

如果传递的 Decimal 小数位数过多，则某些数据库连接器（如 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

将时区感知时间戳存储为 Timezone Naive UTC¶

数据库中的时间戳应始终以与时区无关的方式存储。为大多数数据库，这意味着确保时间戳在 UTC 时区中排在第一位，然后将其存储为 Timezone-naïve（即没有任何与其关联的时区;UTC 被假定为“隐式”时区）。或者，特定于数据库的类型（如 PostgreSQL）的 “TIMESTAMP WITH TIMEZONE“通常因其更丰富的功能而受到青睐;但是，将因为普通 UTC 将适用于所有数据库和驱动程序。当 timezone-intelligent 数据库类型不是一个选项或不是首选类型，则 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 or value.tzinfo.utcoffset(value) is None:
                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 类型¶

注意

从版本 2.0 开始，应该首选行为类似的内置 Uuid 类型。此示例仅作为接收和返回 python 对象的类型装饰器的示例。

接收并返回 Python uuid（）对象。使用 PostgreSQL 时使用 PG UUID 类型，使用 MSSQL 时使用 UNIQUEIDENTIFIER，在其他后端使用 CHAR（32）类型，以字符串化格式存储它们。GUIDHyphens 版本使用 CHAR（36）类型使用连字符而不是十六进制字符串存储值：

from operator import attrgetter
from sqlalchemy.types import TypeDecorator, CHAR
from sqlalchemy.dialects.mssql import UNIQUEIDENTIFIER
from sqlalchemy.dialects.postgresql import UUID
import uuid


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

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

    """

    impl = CHAR
    cache_ok = True

    _default_type = CHAR(32)
    _uuid_as_str = attrgetter("hex")

    def load_dialect_impl(self, dialect):
        if dialect.name == "postgresql":
            return dialect.type_descriptor(UUID())
        elif dialect.name == "mssql":
            return dialect.type_descriptor(UNIQUEIDENTIFIER())
        else:
            return dialect.type_descriptor(self._default_type)

    def process_bind_param(self, value, dialect):
        if value is None or dialect.name in ("postgresql", "mssql"):
            return value
        else:
            if not isinstance(value, uuid.UUID):
                value = uuid.UUID(value)
            return self._uuid_as_str(value)

    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


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

    Uses PostgreSQL's UUID type or MSSQL's UNIQUEIDENTIFIER,
    otherwise uses CHAR(36), storing as stringified uuid values.

    """

    _default_type = CHAR(36)
    _uuid_as_str = str

链接 Python `uuid。UUID` 设置为 ORM 映射的自定义类型¶

当使用带注释的声明式表声明 ORM 映射时 mappings，则上面定义的自定义 GUID 类型可能与 Python uuid 相关联。UUID 数据类型，将其添加到 type 注释映射，通常在 DeclarativeBase 类上定义：

import uuid
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    type_annotation_map = {
        uuid.UUID: GUID,
    }

通过上述配置，ORM 映射了从 Base 可以引用 Python uuid。UUID 将自动使用 GUID：

class MyModel(Base):
    __tablename__ = "my_table"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True)

另请参阅

自定义类型映射

Marshal 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 运算符。LIKE 对 JSON 结构没有意义，但对底层文本表示有意义。要使用 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 提供了一个内置系统，用于根据运算符进行此类类型转换。如果我们想经常使用 LIKE 运算符，并将 JSON 对象解释为字符串，我们可以通过覆盖 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” 等运算符的一种方法。其他应用程序可能希望为对 JSON 对象（如 “LIKE”）没有意义的运算符引发 NotImplementedError，而不是自动强制转换为文本。

应用 SQL 级 Bind/Result 处理¶

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

任何 TypeEngine、UserDefinedType 或 TypeDecorator 子类可以包含 TypeEngine.bind_expression（） 和/或 TypeEngine.column_expression() ，当定义为返回非 None 值时，应返回 ColumnElement 表达式，或者将 bound 参数或列表达式。例如，要构建 Geometry type 将 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 对 bound 参数运行，以便转换传入的值：

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 表达式不会干扰包装表达式的标记。例如，如果我们针对表达式的 label（） 渲染 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+psycopg2://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 定义了一组可用于所有列表达式的固定表达式运算符。其中一些作具有重载 Python 的内置运算符的效果; 此类运算符的示例包括 ColumnOperators.__eq__（） （table.c.somecolumn == 'foo'）， ColumnOperators.__invert__（） （~table.c.flag）和 ColumnOperators.__add__（） （table.c.x + table.c.y）。其他运算符公开为显式方法，例如 ColumnOperators.in_（） （table.c.value.in_（['x'， 'y']））和 ColumnOperators.like（） （table.c.value.like（'%ed%'））.

当需要上面已经提供的方法不直接支持的 SQL 运算符时，生成此运算符的最方便方法是在任何 SQL 表达式对象上使用 Operators.op（） 方法;此方法给定一个字符串，表示要渲染的 SQL 运算符，返回值是接受任意右侧表达式的 Python 可调用对象：

>>> fromsqlalchemyimportcolumn>>> expr=column("x").op(">>")(column("y"))>>> print(expr)x >> y

当使用自定义 SQL 类型时，还有一种方法可以实现如上所述的自定义运算符，这些运算符会自动出现在使用该列类型的任何列表达式上，而无需在每次使用运算符时直接调用 Operators.op（）。

为了实现这一点，SQL 表达式构造会查询关联的 TypeEngine 对象替换为构造来确定内置运算符以及查找可能已调用的新方法。 TypeEngine 定义了一个由 Comparator 类实现的 “比较” 对象，以提供 SQL 运算符的基本行为，并且许多特定类型提供了此类的子实现。用户定义的比较器 实现可以直接构建到特定键入以覆盖或定义新作。下面，我们创建一个 覆盖 ColumnOperators.__add__（） 的 Integer 子类 operator，它反过来使用 Operators.op（） 生成自定义 SQL 本身：

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 属性设置为引用新类，将与 Integer 类型关联的 Comparator 类子类化。

用法：

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

ColumnOperators.__add__（） 的实现由拥有的 SQL 表达式查询，方法是将 Comparator 实例化，并将自身作为 expr 属性。当实现需要引用原始 ColumnElement 时，可以使用此属性 object 直接访问：

from sqlalchemy import Integer


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

添加到 Comparator 的新方法公开在使用动态查找方案拥有 SQL 表达式对象，该方案公开添加到 Comparator 添加到拥有的 ColumnElement 上 expression 构造。例如，要向整数添加 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)

当使用 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)

一元运算也是可能的。例如，要添加 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
            )

使用上述类型：

>>> fromsqlalchemy.sqlimportcolumn>>> print(column("x",MyInteger).factorial())x !

另请参阅

运算符.op（）

TypeEngine.comparator_factory

创建新类型¶

UserDefinedType 类作为简单的基类提供，用于定义全新的数据库类型。使用它来表示 SQLAlchemy 未知的本机数据库类型。如果只需要 Python 转换行为，请改用 TypeDecorator。

对象名称	描述
UserDefinedType （用户定义类型）	用户定义类型的 Base。

类 sqlalchemy.types 中。UserDefinedType（用户定义类型）¶

用户定义类型的 Base。

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

import sqlalchemy.types as types


class MyType(types.UserDefinedType):
    cache_ok = True

    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",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("data", MyType(16)),
)

在大多数情况下，get_col_spec（） 方法将接收一个关键字参数 type_expression，该参数引用正在编译的类型的拥有表达式，例如 Column 或 cast（） 结构。仅当方法在其参数签名中接受关键字参数（例如 **kw）时，才会发送此关键字;Introspection 用于检查此函数，以支持此函数的旧形式。

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

1.4.28 版本中的新功能：广义了ExternalType.cache_ok 标志，使其可用于 TypeDecorator 和 UserDefinedType。

成员

cache_ok、coerce_compared_value（） ensure_kwarg

类签名

类 sqlalchemy.types.UserDefinedType （sqlalchemy.types.ExternalType， sqlalchemy.types.TypeEngineMixin ， sqlalchemy.types.TypeEngine， sqlalchemy.util.langhelpers.EnsureKWArg ）

属性 sqlalchemy.types.UserDefinedType. cache_ok：boolNone = None¶

继承自ExternalType 的 ExternalType.cache_ok属性

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

默认值 None 将发出警告，然后不允许缓存包含此类型的语句。设置为 False 可完全禁止缓存使用此类型的语句，而不会显示警告。当设置为 True 时，对象的类和从其 state 将用作缓存键的一部分。例如，使用 TypeDecorator 中：

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

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

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

>>> MyType(["a", "b", "c"])._static_cache_key
(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

缓存方案将从类型中提取与 __init__（） 方法中的参数名称相对应的属性。在上面，“choices”属性成为缓存键的一部分，但“internal_only”不是，因为没有名为“internal_only”的参数。

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

为了适应引用不可哈希结构（如字典、集合和列表）的数据类型，可以通过将可哈希结构分配给名称与参数名称对应的属性来使这些对象“可缓存”。例如，接受查找值字典的数据类型可能会将其发布为一系列排序的 Tuples。给定一个以前不可缓存的类型为：

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    this is the non-cacheable version, as "self.lookup" is not
    hashable.

    """

    def __init__(self, lookup):
        self.lookup = lookup

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self.lookup" ...

其中 “lookup” 是字典。该类型将无法生成缓存键：

>>> type_ = LookupType({"a": 10, "b": 20})
>>> type_._static_cache_key
<stdin>:1: SAWarning: UserDefinedType LookupType({'a': 10, 'b': 20}) will not
produce a cache key because the ``cache_ok`` flag is not set to True.
Set this flag to True if this type object's state is safe to use
in a cache key, or False to disable this warning.
symbol('no_cache')

如果我们确实设置了这样的缓存键，它将不可用。我们将得到一个包含字典的元组结构，它本身不能用作 “缓存字典” 中的键，例如 SQLAlchemy 的语句缓存，因为 Python 字典是不可哈希的：

>>> # set cache_ok = True
>>> type_.cache_ok = True

>>> # this is the cache key it would generate
>>> key = type_._static_cache_key
>>> key
(<class '__main__.LookupType'>, ('lookup', {'a': 10, 'b': 20}))

>>> # however this key is not hashable, will fail when used with
>>> # SQLAlchemy statement cache
>>> some_cache = {key: "some sql value"}
Traceback (most recent call last): File "<stdin>", line 1,
in <module> TypeError: unhashable type: 'dict'

通过将排序的 Tuples 元组分配给 “.lookup” 属性，可以使类型可缓存：

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    The dictionary is stored both as itself in a private variable,
    and published in a public variable as a sorted tuple of tuples,
    which is hashable and will also return the same value for any
    two equivalent dictionaries.  Note it assumes the keys and
    values of the dictionary are themselves hashable.

    """

    cache_ok = True

    def __init__(self, lookup):
        self._lookup = lookup

        # assume keys/values of "lookup" are hashable; otherwise
        # they would also need to be converted in some way here
        self.lookup = tuple((key, lookup[key]) for key in sorted(lookup))

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self._lookup" ...

在上面，的 LookupType({"a": 10, "b": 20}) 缓存键将是：

>>> LookupType({"a": 10, "b": 20})._static_cache_key
(<class '__main__.LookupType'>, ('lookup', (('a', 10), ('b', 20))))

在 1.4.14 版本加入： - 添加了 cache_ok 标志以允许对 TypeDecorator 类进行一些可配置性缓存。

1.4.28 版本的新Function： - 添加了 ExternalType mixin，它将 cache_ok 标志推广到 TypeDecorator 和 UserDefinedType 类。

另请参阅

SQL 编译缓存

method sqlalchemy.types.UserDefinedType. coerce_compared_value（op：OperatorTypeNone， value： Any）→ TypeEngine[Any]¶

为表达式中的 'coerced' Python 值建议类型。

UserDefinedType 的默认行为与 TypeDecorator 的行为相同;默认情况下，它会返回 self 中，假设比较的值应该被强制转换为与此类型相同。看 TypeDecorator.coerce_compared_value() 了解更多详情。

属性 sqlalchemy.types.UserDefinedType. ensure_kwarg： str = 'get_col_spec'¶

一个正则表达式，指示方法应接受 **kw 参数的方法名称。

该类将扫描与 name 模板匹配的方法，并在必要时对其进行修饰，以确保接受 **kw 参数。

使用自定义类型和反射¶

请务必注意，修改为其他 Python 内行为，包括基于 TypeDecorator 以及其他用户定义的数据类型子类在数据库架构中没有任何表示。当使用反映数据库对象中描述的数据库内省功能时，SQLAlchemy 使用固定映射，该映射将数据库服务器报告的数据类型信息链接到 SQLAlchemy 数据类型对象。例如，如果我们查看 PostgreSQL 架构内部特定数据库列的定义，我们可能会收到字符串 “VARCHAR”。 SQLAlchemy 的 PostgreSQL 方言具有硬编码映射，用于链接字符串名称 “VARCHAR” 添加到 SQLAlchemy VARCHAR 类中，这就是我们发出类似 Table('my_table', m, autoload_with=engine) SQL 的语句时， 其中的 Column 对象将具有 VARCHAR 的实例存在于它里面。

这意味着，如果 Table 对象使用不直接与数据库本机类型名称对应的类型对象，如果我们使用反射针对该数据库表的其他位置的新 MetaData 集合创建新的 Table 对象，它将不具有此数据类型。例如：

>>> fromsqlalchemyimport(...     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 数据类型之上工作，而 LargeBinary 数据类型在 SQLite 上对应于数据库类型 BLOB。在 CREATE TABLE 中，我们看到使用了 BLOB 数据类型。 SQLite 数据库对 PickleType 的 PickleType 的 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 Reflection，因为必要的 表元数据已存在。但是，对于 application 或它们的组合都需要同时使用显式的 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 的 PickleType 中。反射过程将离开此列 项目完好无损：

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

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

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

SQLAlchemy 2.0 文档

SQLAlchemy 2.0 文档

SQLAlchemy 核心

自定义类型¶

覆盖类型编译¶

扩充现有类型¶

TypeDecorator 配方¶

将编码的字符串强制转换为 Unicode¶

四舍五入数字¶

将时区感知时间戳存储为 Timezone Naive UTC¶

与后端无关的 GUID 类型¶

链接 Python `uuid。UUID` 设置为 ORM 映射的自定义类型¶

Marshal JSON 字符串¶

添加可变性¶

处理比较运算¶

应用 SQL 级 Bind/Result 处理¶

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

创建新类型¶

使用自定义类型和反射¶

SQLAlchemy 2.0 文档

SQLAlchemy 2.0 文档

SQLAlchemy 核心

自定义类型¶

覆盖类型编译¶

扩充现有类型¶

TypeDecorator 配方¶

将编码的字符串强制转换为 Unicode¶

四舍五入数字¶

将时区感知时间戳存储为 Timezone Naive UTC¶

与后端无关的 GUID 类型¶

链接 Python uuid。UUID 设置为 ORM 映射的自定义类型¶

Marshal JSON 字符串¶

添加可变性¶

处理比较运算¶

应用 SQL 级 Bind/Result 处理¶

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

创建新类型¶

使用自定义类型和反射¶

链接 Python `uuid。UUID` 设置为 ORM 映射的自定义类型¶