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

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

Часто возникает необходимость принудительно изменить «строковую» версию типа, то есть ту, которая отображается в операторе 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.

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

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

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

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

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

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. Это целесообразно в тех случаях, когда только реляционная база данных содержит определенный набор функций, необходимых для преобразования входящих и исходящих данных между приложением и форматом сохранения. В качестве примера можно привести использование определенных базой данных функций шифрования/дешифрования, а также хранимых процедур для работы с географическими данными. Расширение PostGIS для PostgreSQL включает обширный набор 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://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)
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)
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
            )

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

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

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

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

Работа с пользовательскими типами и отражение¶

Важно отметить, что типы баз данных, которые модифицированы так, чтобы иметь дополнительное поведение в 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)
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 является отраженным и загружает определение столбца «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.