SqlAlchemy 2.0 中文文档（四十二）（4）-阿里云开发者社区

SqlAlchemy 2.0 中文文档（四十二）（3）https://developer.aliyun.com/article/1563014

编组 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”这样的运算符的一种方法。其他应用程序可能希望对于没有与 JSON 对象具有意义的运算符（如“LIKE”）引发NotImplementedError，而不是自动强制转换为文本。

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

关于 Unicode 类型的一个常见困惑是，它只打算在 Python 侧处理 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

四舍五入数值

如果传递的 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 的“带时区的时间戳”这样的数据库特定类型，因为其更丰富的功能；然而，将其存储为纯 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)

另请参阅

自定义类型映射

将 Python `uuid.UUID` 链接到 ORM 映射的自定义类型

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”之类运算符的一种方法。其他应用程序可能希望对于 JSON 对象没有意义的运算符（如“LIKE”）引发NotImplementedError，而不是自动强制转换为文本。

添加可变性

ORM 默认不会检测上述类型的“可变性”——这意味着对值的原地更改不会被检测到，也不会被刷新。如果没有进一步的步骤，您将需要在每个父对象上用新值替换现有值以检测更改：

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

以上限制可能是可以接受的，因为许多应用程序可能不要求一旦创建就对值进行突变。对于那些具有此要求的应用程序，最好使用sqlalchemy.ext.mutable扩展来支持可变性。对于面向字典的 JSON 结构，我们可以这样应用：

json_type = MutableDict.as_mutable(JSONEncodedDict)
class MyClass(Base):
    #  ...
    json_data = Column(json_type)

参见

突变跟踪

处理比较操作

TypeDecorator的默认行为是将任何表达式的“右侧”强制转换为相同类型。对于像 JSON 这样的类型，这意味着任何使用的运算符都必须从 JSON 的角度来看是有意义的。对于某些情况，用户可能希望该类型在某些情况下表现得像 JSON，在其他情况下表现为纯文本。一个例子是，如果一个人希望处理 JSON 类型的 LIKE 运算符。对于 JSON 结构来说，LIKE 没有意义，但对于基础文本表示来说是有意义的。要想在像JSONEncodedDict这样的类型中实现这一点，我们需要使用cast()或type_coerce()将列强制转换为文本形式，然后再尝试使用此运算符：

from sqlalchemy import type_coerce, String
stmt = select(my_table).where(type_coerce(my_table.c.json_data, String).like("%foo%"))

TypeDecorator提供了一个基于运算符的系统，用于处理此类类型转换。如果我们想要频繁地使用 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，而不是自动转换为文本。

SqlAlchemy 2.0 中文文档（四十二）（5）https://developer.aliyun.com/article/1563016

SqlAlchemy 2.0 中文文档（四十二）（4）

编组 JSON 字符串

添加可变性

处理比较操作

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

四舍五入数值

将时区感知时间戳存储为时区无关的 UTC

跨后端 GUID 类型

将 Python `uuid.UUID` 链接到 ORM 映射的自定义类型

将 Python `uuid.UUID` 链接到 ORM 映射的自定义类型

编组 JSON 字符串

添加可变性

处理比较操作

添加可变性

处理比较操作

热门文章

最新文章

相关电子书

SqlAlchemy 2.0 中文文档（四十二）（4）

编组 JSON 字符串

添加可变性

处理比较操作

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

四舍五入数值

将时区感知时间戳存储为时区无关的 UTC

跨后端 GUID 类型

将 Python uuid.UUID 链接到 ORM 映射的自定义类型

将 Python uuid.UUID 链接到 ORM 映射的自定义类型

编组 JSON 字符串

添加可变性

处理比较操作

添加可变性

处理比较操作

热门文章

最新文章

相关电子书

将 Python `uuid.UUID` 链接到 ORM 映射的自定义类型

将 Python `uuid.UUID` 链接到 ORM 映射的自定义类型