自定义类型¶

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

覆盖类型编译¶

一个常见的需求是强制类型的“字符串”版本，即在 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')

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

>>> # 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¶ – 操作的 “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 是可调用的，并且生成的对象将分配给 self.impl 实例属性（因此覆盖同名的类属性）。

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

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

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

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

注意

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

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

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

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

此方法实现了 TypeEngine 契约，用于绑定值的转换，这通常通过 TypeEngine.bind_processor() 方法发生。

注意

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

参数:: dialect¶ – 使用中的 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 调用此方法。返回值是对于此特定操作，35 应使用的任何 SQLAlchemy 类型。

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

指定当使用 == 比较（以及与 != 结合使用的 IS NOT 相同）时，应在表达式级别强制转换为“IS <constant>”的那些 Python 类型。

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

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

自定义 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]¶

返回与 dialect 对应的 TypeEngine 对象。

这是一个最终用户覆盖钩子，可用于根据给定的 dialect 提供不同的类型。 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¶ – 使用中的 Dialect 实例。
coltype¶ – SQLAlchemy 数据类型

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

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

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

1.3.8 版本中的新功能。

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

为此 TypeDecorator 返回一个 dialect 特定的 TypeEngine 实例。

在大多数情况下，这会返回由 self.impl 表示的 TypeEngine 类型的 dialect 适配形式。使用 dialect_impl()。可以通过覆盖 load_dialect_impl() 在此处自定义行为。

TypeDecorator 食谱¶

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

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

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

可以使用 TypeDecorator 来实现可以安全接收 Python 字节串的类型用例，即包含非 ASCII 字符且在 Python 2 中不是 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

将时区感知时间戳存储为时区无关的 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。可以修改为使用 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”这样的运算符的一种方法。其他应用程序可能希望对于诸如“LIKE”之类的对 JSON 对象没有意义的运算符引发 NotImplementedError，而不是自动强制转换为文本。

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

如增强现有类型章节中所见，SQLAlchemy 允许在参数发送到语句时以及从数据库加载结果行时调用 Python 函数，以便在值发送到或从数据库时应用转换。也可以定义 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 属性建立为引用一个新类，该类是 Comparator 类的子类，而 Comparator 类又与 Integer 类型关联。

用法

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

ColumnOperators.__add__() 的实现由拥有的 SQL 表达式查询，方法是将 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')

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

>>> # 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 的另一个实例，则 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 对象中存储为 PickleType。

在实践中，上述基于事件的方法可能会有额外的规则，以便仅影响那些数据类型重要的列，例如表名和可能的列名的查找表，或其他启发式方法，以便准确确定哪些列应该使用 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 映射¶