Пользовательские типы¶

Переопределение компиляции типов¶

Часто возникает необходимость принудительно изменить «строковую» версию типа, то есть ту, которая отображается в операторе CREATE TABLE или другой функции SQL, например CAST. Например, приложение может захотеть принудительно отобразить BINARY для всех платформ, кроме одной, на которой нужно отобразить BLOB. Использование существующего общего типа, в данном случае LargeBinary, является предпочтительным для большинства случаев использования. Но для более точного управления типами, директива компиляции, которая является per-dialect, может быть связана с любым типом:

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import BINARY


@compiles(BINARY, "sqlite")
def compile_binary_sqlite(type_, compiler, **kw):
    return "BLOB"

Приведенный выше код позволяет использовать BINARY, который выдаст строку BINARY против всех бэкендов, кроме SQLite, в этом случае он выдаст BLOB.

Дополнительные примеры см. в разделе Изменение компиляции типов, подразделе Пользовательские SQL-конструкции и расширение компиляции.

Дополнение существующих типов¶

TypeDecorator позволяет создавать пользовательские типы, которые добавляют поведение привязки параметров и обработки результатов к существующему объекту типа. Он используется в тех случаях, когда требуется дополнительная обработка данных в базе данных и/или из базы данных на языке In-Python marshalling.

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

    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 с приведенным выше типом заставит «дату» в правой части также рассматриваться как 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)

Members

cache_ok, operate(), reverse_operate(), __init__(), bind_expression(), bind_processor(), coerce_compared_value(), coerce_to_is_types, column_expression(), comparator_factory, compare_values(), copy(), get_dbapi_type(), literal_processor(), load_dialect_impl(), process_bind_param(), process_literal_param(), process_result_value(), result_processor(), sort_key_function, type_engine()

Классная подпись

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

attribute sqlalchemy.types.TypeDecorator.cache_ok: Optional[bool] = 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: - added the cache_ok flag to allow some configurability of caching for TypeDecorator classes.

Добавлено в версии 1.4.28: - added the ExternalType mixin which generalizes the cache_ok flag to both the TypeDecorator and UserDefinedType classes.

См.также

Кэширование компиляции SQL

class Comparator¶

Comparator, который является специфическим для TypeDecorator.

Определяемые пользователем TypeDecorator классы обычно не должны изменять это.

Классная подпись

класс 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 является вызываемым, а полученный объект назначается атрибуту экземпляра self.impl (тем самым переопределяя одноименный атрибут класса).

Если уровень класса impl не является вызываемым (необычный случай), он будет присвоен тому же атрибуту экземпляра «как есть», игнорируя аргументы, переданные конструктору.

Подклассы могут переопределить его, чтобы полностью настроить генерацию self.impl.

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

Учитывая значение привязки (т.е. экземпляр BindParameter), верните SQL-выражение, которое обычно обертывает заданный параметр.

Примечание

Этот метод вызывается на этапе SQL-компиляции оператора при визуализации строки SQL. Он не обязательно вызывается против конкретных значений, и его не следует путать с методом TypeDecorator.process_bind_param(), который является более типичным методом, обрабатывающим фактическое значение, переданное конкретному параметру во время выполнения оператора.

Подклассы TypeDecorator могут переопределить этот метод, чтобы обеспечить пользовательское поведение выражения связывания для типа. Эта реализация будет заменять реализацию базового типа.

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

Предоставить функцию обработки связанного значения для заданного Dialect.

Это метод, выполняющий контракт TypeEngine для преобразования связанного значения, которое обычно происходит через метод TypeEngine.bind_processor().

Примечание

Определяемые пользователем подклассы TypeDecorator должны не реализовывать этот метод, и вместо этого должны реализовывать TypeDecorator.process_bind_param(), чтобы сохранялась «внутренняя» обработка, предоставляемая реализующим типом.

Параметры:

dialect – Используемый диалектный экземпляр.

method sqlalchemy.types.TypeDecorator.coerce_compared_value(op: Optional[OperatorType], value: Any) → Any¶

Предложите тип для «принудительного» значения Python в выражении.

По умолчанию возвращает self. Этот метод вызывается системой выражений, когда объект, использующий этот тип, находится в левой или правой части выражения против обычного объекта Python, которому еще не присвоен тип SQLAlchemy:

expr = table.c.somecolumn + 35

Если somecolumn использует этот тип, то этот метод будет вызван со значением operator.add и 35. Возвращаемое значение - это тип SQLAlchemy, который должен быть использован для 35 для данной конкретной операции.

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

Укажите те типы Python, которые должны принудительно приводиться на уровне выражения к «IS <constant>» при сравнении с помощью == (и то же самое для IS NOT в сочетании с !=).

Для большинства типов SQLAlchemy это включает NoneType, а также bool.

TypeDecorator модифицирует этот список, включив в него только NoneType, поскольку реализации типизаторов, работающие с булевыми типами, встречаются часто.

Пользовательские классы TypeDecorator могут переопределить этот атрибут, чтобы вернуть пустой кортеж, в этом случае никакие значения не будут приводиться к константам.

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

Учитывая выражение столбца 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¶

Учитывая два значения, сравните их на равенство.

По умолчанию это вызывает TypeEngine.compare_values() из базового «impl», который, в свою очередь, обычно использует оператор равенства Python ==.

Эта функция используется ORM для сравнения исходного загруженного значения с перехваченным «измененным» значением, чтобы определить, произошло ли чистое изменение.

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

Создайте копию данного экземпляра TypeDecorator.

Это неглубокая копия, которая предоставляется для выполнения части контракта TypeEngine. Обычно его не нужно переопределять, если только определяемый пользователем TypeDecorator не имеет локального состояния, которое должно быть глубоко скопировано.

method sqlalchemy.types.TypeDecorator.get_dbapi_type(dbapi: module) → Optional[Any]¶

Возвращает объект типа DBAPI, представленный данным TypeDecorator.

По умолчанию это вызывает TypeEngine.get_dbapi_type() из базового «impl».

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

Предоставляет функцию обработки литерала для заданного 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: Optional[_T], dialect: Dialect) → Any¶

Получение связанного значения параметра для преобразования.

Пользовательские подклассы TypeDecorator должны переопределить этот метод, чтобы обеспечить пользовательское поведение для входящих значений данных. Этот метод вызывается во время выполнения оператора и ему передается буквальное значение данных Python, которое должно быть связано со связанным параметром в операторе.

Операция может быть любой для выполнения пользовательского поведения, например, преобразования или сериализации данных. Это также может быть использовано в качестве крючка для проверки логики.

Параметры:

• value – Данные для работы, любого типа, ожидаемого этим методом в подклассе. Может быть None.

• dialect – используемый Dialect.

См.также

Дополнение существующих типов

TypeDecorator.process_result_value()

method sqlalchemy.types.TypeDecorator.process_literal_param(value: Optional[_T], dialect: Dialect) → str¶

Получение буквального значения параметра для отображения в строке внутри оператора.

Примечание

Этот метод вызывается на этапе SQL-компиляции оператора при визуализации строки SQL. В отличие от других методов компиляции SQL, ему передается конкретное значение Python, которое должно быть преобразовано в строку. Однако его не следует путать с методом TypeDecorator.process_bind_param(), который является более типичным методом, обрабатывающим фактическое значение, переданное конкретному параметру во время выполнения оператора.

Пользовательские подклассы TypeDecorator должны переопределить этот метод, чтобы обеспечить пользовательское поведение для входящих значений данных, которые в особом случае отображаются как литералы.

Возвращенная строка будет преобразована в выходную строку.

method sqlalchemy.types.TypeDecorator.process_result_value(value: Optional[Any], dialect: Dialect) → Optional[_T]¶

Получите значение столбца результирующей строки для преобразования.

Пользовательские подклассы TypeDecorator должны переопределить этот метод, чтобы обеспечить пользовательское поведение для значений данных, получаемых в строках результатов, поступающих из базы данных. Этот метод вызывается в время получения результата и ему передается буквальное значение данных Python, которое извлекается из строки результата базы данных.

Операция может быть любой для выполнения пользовательского поведения, например, преобразования или десериализации данных.

Параметры:

• value – Данные для работы, любого типа, ожидаемого этим методом в подклассе. Может быть None.

• dialect – используемый Dialect.

См.также

Дополнение существующих типов

TypeDecorator.process_bind_param()

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

Предоставить функцию обработки значения результата для заданного Dialect.

Это метод, выполняющий контракт TypeEngine для преобразования связанного значения, которое обычно происходит через метод TypeEngine.result_processor().

Примечание

Определяемые пользователем подклассы TypeDecorator должны не реализовывать этот метод, и вместо этого должны реализовывать TypeDecorator.process_result_value(), чтобы сохранялась «внутренняя» обработка, предоставляемая реализующим типом.

Параметры:

• dialect – Используемый диалектный экземпляр.

• coltype – Тип данных SQLAlchemy

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

Функция сортировки, которая может быть передана как ключ к sorted.

Значение по умолчанию None указывает на то, что значения, хранимые этим типом, являются самосортирующимися.

Добавлено в версии 1.3.8.

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

Возвращает специфичный для диалекта TypeEngine экземпляр для данного TypeDecorator.

В большинстве случаев это возвращает адаптированную к диалекту форму типа TypeEngine, представленного self.impl. Используется dialect_impl(). Поведение здесь может быть настроено путем переопределения load_dialect_impl().

ТипДекоратор Рецепты¶

Далее следуют несколько ключевых рецептов 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

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:
                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 sqlalchemy.types import TypeDecorator, CHAR
from sqlalchemy.dialects.postgresql import UUID
import uuid


class GUID(TypeDecorator):
    """Platform-independent GUID type.

    Uses PostgreSQL's UUID type, otherwise uses
    CHAR(32), storing as stringified hex values.

    """

    impl = CHAR
    cache_ok = True

    def load_dialect_impl(self, dialect):
        if dialect.name == "postgresql":
            return dialect.type_descriptor(UUID())
        else:
            return dialect.type_descriptor(CHAR(32))

    def process_bind_param(self, value, dialect):
        if value is None:
            return value
        elif dialect.name == "postgresql":
            return str(value)
        else:
            if not isinstance(value, uuid.UUID):
                return "%.32x" % uuid.UUID(value).int
            else:
                # hexstring
                return "%.32x" % value.int

    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

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

Применение привязки/обработки результатов на уровне 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 применяется к предложению columns, чтобы возвращаемое значение прогонялось через функцию перед передачей в набор результатов, а 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 не вмешивается в маркировку обернутого выражения. Например, если мы поместили select() против label() нашего выражения, метка строки перемещается за пределы обернутого выражения:

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, который не поддерживается непосредственно вышеуказанными методами, наиболее целесообразным способом получения этого оператора является использование метода Operators.op() на любом объекте выражения SQL; этому методу передается строка, представляющая оператор SQL для визуализации, а возвращаемым значением является вызываемый объект Python, который принимает любое произвольное выражение правой части:

>>> from sqlalchemy import column
>>> expr = column("x").op(">>")(column("y"))
>>> print(expr)
{printsql}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, связанного с типом Integer.

Использование:

>>> sometable = Table("sometable", metadata, Column("data", MyInt))
>>> print(sometable.c.data + 5)
{printsql}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))
{printsql}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
            )

Используя вышеуказанный тип:

>>> from sqlalchemy.sql import column
>>> print(column("x", MyInteger).factorial())
{printsql}x !

Создание новых типов¶

Класс UserDefinedType предоставляется как простой базовый класс для определения совершенно новых типов баз данных. Используйте его для представления собственных типов баз данных, не известных SQLAlchemy. Если требуется только поведение трансляции Python, используйте TypeDecorator вместо него.

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.

Members

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: Optional[bool] = 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: - added the cache_ok flag to allow some configurability of caching for TypeDecorator classes.

Добавлено в версии 1.4.28: - added the ExternalType mixin which generalizes the cache_ok flag to both the TypeDecorator and UserDefinedType classes.

См.также

Кэширование компиляции SQL

method sqlalchemy.types.UserDefinedType.coerce_compared_value(op: Optional[OperatorType], 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". Диалект PostgreSQL SQLAlchemy имеет жестко закодированное отображение, которое связывает строковое имя "VARCHAR" с классом SQLAlchemy VARCHAR, и поэтому, когда мы выдаем оператор типа Table('my_table', m, autoload_with=engine), объект Column внутри него будет иметь экземпляр VARCHAR.

Из этого следует, что если объект Table использует объекты типа, которые не соответствуют непосредственно имени типа, присущего базе данных, если мы создадим новый объект Table против новой коллекции MetaData для этой таблицы базы данных в другом месте, используя отражение, он не будет иметь этот тип данных. Например:

>>> from sqlalchemy import (
...     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)
{execsql}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)
{execsql}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 является отраженным и загружает определение столбца «id» из базы данных SQLite. Но для столбца «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.