自定义类型¶

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

覆盖类型编译¶

一个常见的需求是强制类型“字符串”版本（即在 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，而在 SQLite 中则会生成 BLOB。

请参阅更改类型的编译部分（自定义 SQL 结构和编译扩展的一个子部分），了解其他示例。

扩展现有类型¶

TypeDecorator 允许创建自定义类型，这些自定义类型将绑定参数和结果处理行为添加到现有类型对象中。当需要对数据进行额外的 Python 内封送处理（进出数据库）时，使用它。

注意

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

对象名称	描述
TypeDecorator	允许创建类型，这些类型将向现有类型添加额外的功能。

class sqlalchemy.types.TypeDecorator¶

允许创建类型，这些类型将向现有类型添加额外的功能。

与直接子类化 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

    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() 方法覆盖此行为，该方法返回一个用于表达式值的类型。下面我们将它设置为整数类型将被视为 Integer，任何其他值都被视为日期类型，并将被视为 MyEpochType。

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

警告

请注意，coerce_compared_value 的行为默认情况下不会从基类型的行为继承。如果 TypeDecorator 正在增强需要针对特定类型的运算符进行特殊逻辑的类型，则必须覆盖此方法。一个关键示例是装饰 JSON 和 JSONB 类型时；为了处理诸如索引操作之类的运算符，应使用 TypeEngine.coerce_compared_value() 的默认规则。

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)

类签名

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

attribute sqlalchemy.types.TypeDecorator.cache_ok: bool | None = None¶

继承自 ExternalType.cache_ok 属性的 ExternalType

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

默认值 None 将发出警告，然后不允许缓存包含此类型的语句。设置为 False 以完全禁用使用此类型的语句被缓存（不带警告）。设置为 True 时，对象的类及其状态中选定的元素将用作缓存键的一部分。例如，使用 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。

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

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'

可以通过将排序的元组列表分配给“.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 中的新功能：- 添加了 ExternalType 混合，它将 cache_ok 标志推广到 TypeDecorator 和 UserDefinedType 类。

另请参阅

SQL 编译缓存

class Comparator¶

特定于 TypeDecorator 的 Comparator。

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

类签名

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

method sqlalchemy.types.TypeDecorator.Comparator.operate(op: OperatorType, *other: Any, **kwargs: Any) → ColumnElement[_CT]¶

对参数进行操作。

这是最低级别的操作，默认情况下会引发 NotImplementedError。

在子类中覆盖此方法可以将通用行为应用于所有操作。例如，覆盖 ColumnOperators 以将 func.lower() 应用于左右两侧

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

参数：:

op¶ – 操作可调用对象。
*other¶ – 操作的“另一方”。对于大多数操作，它将是一个单一的标量。
**kwargs¶ – 修饰符。这些可能由特殊的运算符传递，例如 ColumnOperators.contains()。

method sqlalchemy.types.TypeDecorator.Comparator.reverse_operate(op: OperatorType, other: Any, **kwargs: Any) → ColumnElement[_CT]¶

对参数进行反向操作。

用法与 operate() 相同。

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

构造 TypeDecorator。

这里发送的参数将传递给分配给 impl 类级别属性的类的构造函数，假设 impl 是一个可调用对象，并且生成的

如果类级别 impl 不是可调用对象（不寻常的情况），它将被分配给相同的实例属性“原样”，忽略传递

子类可以覆盖此方法来完全自定义 self.impl 的生成。

method sqlalchemy.types.TypeDecorator.bind_expression(bindparam: BindParameter[_T]) → ColumnElement[_T] | None¶

给定一个绑定值（即 BindParameter 实例），返回一个 SQL 表达式，它通常会

注意

此方法在语句的**SQL 编译**阶段调用，在渲染 SQL 字符串时调用。它**不**一定针对特定值调用，

TypeDecorator 的子类可以覆盖此方法以提供类型自定义绑定表达式行为。此实现将

method sqlalchemy.types.TypeDecorator.bind_processor(dialect: Dialect) → _BindProcessorType[_T] | None¶

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

这是实现 TypeEngine 绑定值转换合同的方法，该转换通常通过

注意

TypeDecorator 的用户定义子类**不应**实现此方法，而应实现

参数：:: dialect¶ – 正在使用的方言实例。

method sqlalchemy.types.TypeDecorator.coerce_compared_value(op: OperatorType | None, value: Any) → Any¶

建议在表达式中“强制”Python 值的类型。

默认情况下，返回 self。当使用此类型的对象在表达式中与尚未分配 SQLAlchemy 类型的普通 Python 对象的

expr = table.c.somecolumn + 35

在上面，如果 somecolumn 使用此类型，此方法将使用值 operator.add 和 35 调用。

attribute sqlalchemy.types.TypeDecorator.coerce_to_is_types: Sequence[Type[Any]] = (<class 'NoneType'>,)¶

指定那些应该在表达式级别强制转换为“IS <constant>”的 Python 类型，当使用 == 比较时（以及

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

TypeDecorator 修改此列表，仅包含 NoneType，因为处理布尔类型的

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

method sqlalchemy.types.TypeDecorator.column_expression(column: ColumnElement[_T]) → ColumnElement[_T] | None¶

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

注意

此方法在语句的 **SQL 编译** 阶段调用，用于渲染 SQL 字符串。它**不会**针对特定值调用，不应与 TypeDecorator.process_result_value() 方法混淆，该方法是更典型的方法，用于在语句执行时间之后处理结果行中返回的实际值。

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

有关方法使用的完整描述，请参见 TypeEngine.column_expression() 的说明。

attribute sqlalchemy.types.TypeDecorator.comparator_factory: _ComparatorFactory[Any]¶

一个 Comparator 类，它将应用于拥有 ColumnElement 对象执行的操作。

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

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

method sqlalchemy.types.TypeDecorator.compare_values(x: Any, y: Any) → bool¶

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

默认情况下，这会调用底层“impl”的 TypeEngine.compare_values()，该方法通常使用 Python 等于运算符 ==。

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

method sqlalchemy.types.TypeDecorator.copy(**kw: Any) → Self¶

生成此 TypeDecorator 实例的副本。

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

method sqlalchemy.types.TypeDecorator.get_dbapi_type(dbapi: module) → Any | None¶

返回此 TypeDecorator 代表的 DBAPI 类型对象。

默认情况下，这会调用底层“impl”的 TypeEngine.get_dbapi_type()。

method sqlalchemy.types.TypeDecorator.literal_processor(dialect: Dialect) → _LiteralProcessorType[_T] | None¶

为给定的 Dialect 提供一个文字处理函数。

这是实现 TypeEngine 合同以进行文字值转换的方法，该转换通常通过 TypeEngine.literal_processor() 方法完成。

注意

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

method sqlalchemy.types.TypeDecorator.load_dialect_impl(dialect: Dialect) → TypeEngine[Any]¶

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

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

默认情况下返回 self.impl。

method sqlalchemy.types.TypeDecorator.process_bind_param(value: _T | None, dialect: Dialect) → Any¶

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

TypeDecorator 的自定义子类应该覆盖此方法以提供对传入数据值的自定义行为。此方法在 **语句执行时** 被调用，并传递给定于语句中绑定参数的文字 Python 数据值。

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

参数：:

value¶ – 要操作的数据，可以是子类中此方法预期的任何类型。可以是 None。
dialect¶ – 正在使用的 Dialect。

另请参阅

扩展现有类型

TypeDecorator.process_result_value()

method sqlalchemy.types.TypeDecorator.process_literal_param(value: _T | None, dialect: Dialect) → str¶

接收要直接渲染到语句中的文字参数值。

注意

此方法在语句的 **SQL 编译** 阶段被调用，用于渲染 SQL 字符串。与其他 SQL 编译方法不同，它传递了一个特定的 Python 值来渲染为字符串。但是它不应与 TypeDecorator.process_bind_param() 方法混淆，后者是在语句执行时处理传递给特定参数的实际值的最典型方法。

TypeDecorator 的自定义子类应该覆盖此方法以提供对传入数据值的自定义行为，这些数据值处于作为文字渲染的特殊情况下。

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

method sqlalchemy.types.TypeDecorator.process_result_value(value: Any | None, dialect: Dialect) → _T | None¶

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

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

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

参数：:

value¶ – 要操作的数据，可以是子类中此方法预期的任何类型。可以是 None。
dialect¶ – 正在使用的 Dialect。

另请参阅

扩展现有类型

TypeDecorator.process_bind_param()

method sqlalchemy.types.TypeDecorator.result_processor(dialect: Dialect, coltype: Any) → _ResultProcessorType[_T] | None¶

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

这是履行 TypeEngine 绑定值转换合同的方法，通常通过 TypeEngine.result_processor() 方法进行。

注意

TypeDecorator 的用户定义子类 **不应** 实现此方法，而应该实现 TypeDecorator.process_result_value()，以便由实现类型提供的“内部”处理得以保留。

参数：:

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

attribute sqlalchemy.types.TypeDecorator.sort_key_function: Callable[[Any], Any] | None¶

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

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

1.3.8 版本中的新功能。

method sqlalchemy.types.TypeDecorator.type_engine(dialect: Dialect) → TypeEngine[Any]¶

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

在大多数情况下，它会返回由 self.impl 表示的 TypeEngine 类型的特定于方言的格式。使用 dialect_impl()。可以通过覆盖 load_dialect_impl() 来自定义此处的行为。

TypeDecorator 食谱¶

以下是一些关键的 TypeDecorator 食谱。

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

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

可以使用 TypeDecorator 来实现一个能够安全接收 Python 字节字符串的类型，即包含非 ASCII 字符并且在 Python 2 中不是 u'' 对象的字符串，该类型根据需要进行强制转换。

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 的连接器）如果传递的 Decimal 小数位数过多，就会出现错误。以下是一个将小数位数向下舍入的方案。

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 是“隐式”时区）。或者，使用 PostgreSQL 的“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 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 映射时，可以使用上述自定义的 GUID 类型与 Python uuid.UUID 数据类型关联，方法是将其添加到类型注解映射中，该映射通常在 DeclarativeBase 类上定义。

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


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

使用上述配置，扩展自 Base 的 ORM 映射类可以在注解中引用 Python uuid.UUID，这将自动使用 GUID。

class MyModel(Base):
    __tablename__ = "my_table"

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

另请参阅

自定义类型映射

编组 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 级别的绑定/结果处理¶

如增强现有类型部分所述，SQLAlchemy 允许在将参数发送到语句时以及从数据库加载结果行时调用 Python 函数，以对发送到或从数据库的值进行转换。也可以定义 SQL 级别的转换。这里的原因是，当只有关系数据库包含在应用程序和持久化格式之间对传入数据和传出数据进行强制转换所需的特定函数序列时，可以使用 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 应用于列子句，以便在传递到结果集之前，返回值通过该函数运行，而 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 表达式不会干扰包装表达式的标记。例如，如果我们针对我们表达式的 label() 渲染了一个 select()，则字符串标签将移动到包装表达式的外部。

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 运算符的基本行为，并且许多特定类型提供了此类的自己的子实现。用户定义的 Comparator 实现可以直接构建到特定类型的简单子类中，以便覆盖或定义新的操作。下面，我们创建一个 Integer 子类，该子类覆盖了 ColumnOperators.__add__() 运算符，该运算符依次使用 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

拥有 SQL 表达式的 ColumnOperators.__add__() 实现通过将 Comparator 实例化为自身作为 expr 属性来查询。当实现需要直接引用原始的 ColumnElement 对象时，可以使用此属性。

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 表达式构造中。例如，要为整数添加一个 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 !

另请参阅

Operators.op()

TypeEngine.comparator_factory

创建新类型¶

提供 UserDefinedType 类作为定义全新数据库类型的简单基类。使用此类来表示 SQLAlchemy 未知的原生数据库类型。如果只需要 Python 翻译行为，请改用 TypeDecorator。

对象名称	描述
UserDefinedType	用户自定义类型的基类。

class sqlalchemy.types.UserDefinedType¶

用户自定义类型的基类。

这应该是新类型的基础。注意，对于大多数情况，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）时才会发送此关键字；内省用于检查这一点，以便支持此函数的旧版形式。

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

类签名

class sqlalchemy.types.UserDefinedType (sqlalchemy.types.ExternalType, sqlalchemy.types.TypeEngineMixin, sqlalchemy.types.TypeEngine, sqlalchemy.util.langhelpers.EnsureKWArg)

attribute sqlalchemy.types.UserDefinedType.cache_ok: bool | None = None¶

继承自 ExternalType.cache_ok 属性的 ExternalType

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

默认值 None 将发出警告，然后不允许缓存包含此类型的语句。设置为 False 以完全禁用使用此类型的语句被缓存（不带警告）。设置为 True 时，对象的类及其状态中选定的元素将用作缓存键的一部分。例如，使用 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。

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

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'

可以通过将排序的元组列表分配给“.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 中的新功能：- 添加了 ExternalType 混合，它将 cache_ok 标志推广到 TypeDecorator 和 UserDefinedType 类。

另请参阅

SQL 编译缓存

method sqlalchemy.types.UserDefinedType.coerce_compared_value(op: OperatorType | None, value: Any) → TypeEngine[Any]¶

建议在表达式中“强制”Python 值的类型。

UserDefinedType 的默认行为与 TypeDecorator 相同；默认情况下，它返回 self，假设应该将比较的值强制转换为与该类型相同的类型。有关更多详细信息，请参见 TypeDecorator.coerce_compared_value()。

attribute sqlalchemy.types.UserDefinedType.ensure_kwarg: str = 'get_col_spec'¶

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

该类将扫描匹配名称模板的方法，并在必要时装饰它们，以确保接受 **kw 参数。

使用自定义类型和反射¶

需要注意的是，修改为具有附加 Python 内行为的数据库类型，包括基于 TypeDecorator 的类型以及数据类型的其他用户定义的子类，在数据库模式中没有表示。当使用在反射数据库对象中描述的数据库内省功能时，SQLAlchemy 使用一个固定映射，将数据库服务器报告的数据类型信息链接到 SQLAlchemy 数据类型对象。例如，如果我们在 PostgreSQL 模式中查看特定数据库列的定义，我们可能会接收回字符串 "VARCHAR"。SQLAlchemy 的 PostgreSQL 方言有一个硬编码映射，将字符串名称 "VARCHAR" 链接到 SQLAlchemy VARCHAR 类，这就是当我们发出类似于 Table('my_table', m, autoload_with=engine) 的语句时，其中的 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数据类型之上工作，在 SQLite 中对应于数据库类型BLOB。在 CREATE TABLE 中，我们看到使用了BLOB数据类型。SQLite 数据库对我们使用的PickleType一无所知。

如果我们查看my_table.c.data.type的数据类型，由于这是一个我们直接创建的 Python 对象，它就是PickleType

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

但是，如果我们使用反射创建Table的另一个实例，我们创建的 SQLite 数据库中没有表示使用PickleType；我们得到的是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 级数据类型，以及从数据库中反映出来的设置其Column对象的Table对象，而这些对象仍然需要展现自定义数据类型的额外 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”列，我们用包含我们期望的 Python 内数据类型、PickleType的显式Column定义覆盖了反射对象。反射过程将保留此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()

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

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

SQLAlchemy 2.0 文档

SQLAlchemy 2.0 文档

SQLAlchemy Core

项目版本

自定义类型¶

覆盖类型编译¶

扩展现有类型¶

TypeDecorator 食谱¶

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

舍入数值¶

将带时区的时间戳存储为无时区 UTC 时间¶

与后端无关的 GUID 类型¶

将 Python `uuid.UUID` 与 ORM 映射的自定义类型关联¶

编组 JSON 字符串¶

添加可变性¶

处理比较运算¶

应用 SQL 级别的绑定/结果处理¶

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

创建新类型¶

使用自定义类型和反射¶

SQLAlchemy 2.0 文档

SQLAlchemy 2.0 文档

SQLAlchemy Core

项目版本

自定义类型¶

覆盖类型编译¶

扩展现有类型¶

TypeDecorator 食谱¶

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

舍入数值¶

将带时区的时间戳存储为无时区 UTC 时间¶

与后端无关的 GUID 类型¶

将 Python uuid.UUID 与 ORM 映射的自定义类型关联¶

编组 JSON 字符串¶

添加可变性¶

处理比较运算¶

应用 SQL 级别的绑定/结果处理¶

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

创建新类型¶

使用自定义类型和反射¶

将 Python `uuid.UUID` 与 ORM 映射的自定义类型关联¶