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

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import BINARY
@compiles(BINARY, "sqlite")
def compile_binary_sqlite(type_, compiler, **kw):
    return "BLOB"

class sqlalchemy.types.TypeDecorator

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)

mytable.c.somecol + datetime.date(2009, 5, 15)

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)

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

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)

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)

attribute cache_ok: bool | None = None

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

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" ...

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

>>> # 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'

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})._static_cache_key
(<class '__main__.LookupType'>, ('lookup', (('a', 10), ('b', 20))))

class Comparator

method operate(op: OperatorType, *other: Any, **kwargs: Any) → ColumnElement[_CT]

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

method reverse_operate(op: OperatorType, other: Any, **kwargs: Any) → ColumnElement[_CT]

method __init__(*args: Any, **kwargs: Any)

method bind_expression(bindparam: BindParameter[_T]) → ColumnElement[_T] | None

method bind_processor(dialect: Dialect) → _BindProcessorType[_T] | None

method coerce_compared_value(op: OperatorType | None, value: Any) → Any

expr = table.c.somecolumn + 35

attribute coerce_to_is_types: Sequence[Type[Any]] = (<class 'NoneType'>,)

method column_expression(column: ColumnElement[_T]) → ColumnElement[_T] | None

attribute comparator_factory: _ComparatorFactory[Any]

method compare_values(x: Any, y: Any) → bool

method copy(**kw: Any) → Self

method get_dbapi_type(dbapi: module) → Any | None

method literal_processor(dialect: Dialect) → _LiteralProcessorType[_T] | None

method load_dialect_impl(dialect: Dialect) → TypeEngine[Any]

method process_bind_param(value: _T | None, dialect: Dialect) → Any

method process_literal_param(value: _T | None, dialect: Dialect) → str

method process_result_value(value: Any | None, dialect: Dialect) → _T | None

method result_processor(dialect: Dialect, coltype: Any) → _ResultProcessorType[_T] | None

attribute sort_key_function: Callable[[Any], Any] | None

method type_engine(dialect: Dialect) → TypeEngine[Any]

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

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

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

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

import uuid
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
    type_annotation_map = {
        uuid.UUID: GUID,
    }

class MyModel(Base):
    __tablename__ = "my_table"
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True)

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

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

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

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

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