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 映射的自定义类型
在使用 注释声明的声明性表 映射来声明 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”之类运算符的一种方法。其他应用程序可能希望对于 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