自定义类型¶

Custom Types

中文

存在多种方法可以重新定义现有类型的行为以及提供新的类型的行为。

英文

A variety of methods exist to redefine the behavior of existing types as well as to provide new ones.

重写类型编译¶

Overriding Type Compilation

中文

一个常见的需求是强制类型的“字符串”版本发生改变，也就是说，强制其在 CREATE TABLE 语句或其他 SQL 函数（如 CAST）中渲染的文本。例如，一个应用可能希望在所有平台上都渲染为 BINARY，但在某一个平台上则希望渲染为 BLOB。对于大多数使用场景，推荐使用现有的通用类型，如 LargeBinary。但如果需要更精确地控制类型，可以为任意类型设置一个与方言相关的编译指令，如下:

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，在 SQLite 上会生成 BLOB。

更多示例请参阅章节更改类型的编译，该章节属于自定义 SQL 构造和编译扩展的子章节。

英文

A frequent need is to force the “string” version of a type, that is the one rendered in a CREATE TABLE statement or other SQL function like CAST, to be changed. For example, an application may want to force the rendering of BINARY for all platforms except for one, in which is wants BLOB to be rendered. Usage of an existing generic type, in this case LargeBinary, is preferred for most use cases. But to control types more accurately, a compilation directive that is per-dialect can be associated with any type:

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


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

The above code allows the usage of BINARY, which will produce the string BINARY against all backends except SQLite, in which case it will produce BLOB.

See the section 更改类型的编译, a subsection of 自定义 SQL 构造和编译扩展, for additional examples.

增强现有类型¶

Augmenting Existing Types

中文

TypeDecorator 类允许我们基于现有类型对象创建自定义类型，这些类型可添加绑定参数（bind-parameter）处理和结果处理（result-processing）行为。当数据在 Python 与数据库之间传输时需要额外的转换处理（即术语 marshalling）时，可使用此类。

TypeDecorator 的绑定与结果处理是 附加于 所托管类型已经执行的处理逻辑之上的。 SQLAlchemy 针对每种 DBAPI 都自定义了这些基础类型的处理逻辑，并已涵盖该 DBAPI 所需的全部转换。虽然可以通过子类化来替换这些底层处理行为，但实际中并无必要，且 SQLAlchemy 已不再将其作为公开支持的使用方式。

TypeDecorator 可用于在数据写入和读取数据库时，提供一致的值转换方式。在 ORM 使用中，若要将用户输入的数据从任意格式转换为模型所需格式，可使用 validates() 装饰器实现类似的效果。当转换逻辑依赖于具体的业务需求，且并非一种通用数据类型时，使用验证器装饰器可能更合适。

class sqlalchemy.types.TypeDecorator¶

Allows the creation of types which add additional functionality to an existing type.

This method is preferred to direct subclassing of SQLAlchemy’s built-in types as it ensures that all required functionality of the underlying type is kept in place.

Typical usage:

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)

The class-level impl attribute is required, and can reference any TypeEngine class. Alternatively, the load_dialect_impl() method can be used to provide different type classes based on the dialect given; in this case, the impl variable can reference TypeEngine as a placeholder.

The TypeDecorator.cache_ok class-level flag indicates if this custom TypeDecorator is safe to be used as part of a cache key. This flag defaults to None which will initially generate a warning when the SQL compiler attempts to generate a cache key for a statement that uses this type. If the TypeDecorator is not guaranteed to produce the same bind/result behavior and SQL generation every time, this flag should be set to False; otherwise if the class produces the same behavior each time, it may be set to True. See TypeDecorator.cache_ok for further notes on how this works.

Types that receive a Python type that isn’t similar to the ultimate type used may want to define the TypeDecorator.coerce_compared_value() method. This is used to give the expression system a hint when coercing Python objects into bind parameters within expressions. Consider this expression:

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

Above, if “somecol” is an Integer variant, it makes sense that we’re doing date arithmetic, where above is usually interpreted by databases as adding a number of days to the given date. The expression system does the right thing by not attempting to coerce the “date()” value into an integer-oriented bind parameter.

However, in the case of TypeDecorator, we are usually changing an incoming Python type to something new - TypeDecorator by default will “coerce” the non-typed side to be the same type as itself. Such as below, we define an “epoch” type that stores a date value as an integer:

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)

Our expression of somecol + date with the above type will coerce the “date” on the right side to also be treated as MyEpochType.

This behavior can be overridden via the TypeDecorator.coerce_compared_value() method, which returns a type that should be used for the value of the expression. Below we set it such that an integer value will be treated as an Integer, and any other value is assumed to be a date and will be treated as a MyEpochType:

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

警告

Note that the behavior of coerce_compared_value is not inherited by default from that of the base type. If the TypeDecorator is augmenting a type that requires special logic for certain types of operators, this method must be overridden. A key example is when decorating the JSON and JSONB types; the default rules of TypeEngine.coerce_compared_value() should be used in order to deal with operators like index operations:

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)

Without the above step, index operations such as mycol['foo'] will cause the index value 'foo' to be JSON encoded.

Similarly, when working with the ARRAY datatype, the type coercion for index operations (e.g. mycol[5]) is also handled by TypeDecorator.coerce_compared_value(), where again a simple override is sufficient unless special rules are needed for particular operators:

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)

Class signature

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

class Comparator¶

A Comparator that is specific to TypeDecorator.

User-defined TypeDecorator classes should not typically need to modify this.

Class signature

class sqlalchemy.types.TypeDecorator.Comparator (sqlalchemy.types.Comparator)

method sqlalchemy.types.TypeDecorator.Comparator.operate(op: OperatorType, *other: Any, **kwargs: Any) → ColumnElement[_CT]¶

Operate on an argument.

This is the lowest level of operation, raises NotImplementedError by default.

Overriding this on a subclass can allow common behavior to be applied to all operations. For example, overriding ColumnOperators to apply func.lower() to the left and right side:

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

参数:

op¶ – Operator callable.
*other¶ – the ‘other’ side of the operation. Will be a single scalar for most operations.
**kwargs¶ – modifiers. These may be passed by special operators such as ColumnOperators.contains().

method sqlalchemy.types.TypeDecorator.Comparator.reverse_operate(op: OperatorType, other: Any, **kwargs: Any) → ColumnElement[_CT]¶

Reverse operate on an argument.

Usage is the same as operate().

method sqlalchemy.types.TypeDecorator.__init__(*args: Any, **kwargs: Any)¶

Construct a TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute ‘as-is’, ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

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

Given a bind value (i.e. a BindParameter instance), return a SQL expression which will typically wrap the given parameter.

备注

This method is called during the SQL compilation phase of a statement, when rendering a SQL string. It is not necessarily called against specific values, and should not be confused with the TypeDecorator.process_bind_param() method, which is the more typical method that processes the actual value passed to a particular parameter at statement execution time.

Subclasses of TypeDecorator can override this method to provide custom bind expression behavior for the type. This implementation will replace that of the underlying implementation type.

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

Provide a bound value processing function for the given Dialect.

This is the method that fulfills the TypeEngine contract for bound value conversion which normally occurs via the TypeEngine.bind_processor() method.

备注

User-defined subclasses of TypeDecorator should not implement this method, and should instead implement TypeDecorator.process_bind_param() so that the “inner” processing provided by the implementing type is maintained.

参数:: dialect¶ – Dialect instance in use.

method sqlalchemy.types.TypeDecorator.coerce_compared_value(op: OperatorType | None, value: Any) → Any¶

Suggest a type for a ‘coerced’ Python value in an expression.

By default, returns self. This method is called by the expression system when an object using this type is on the left or right side of an expression against a plain Python object which does not yet have a SQLAlchemy type assigned:

expr = table.c.somecolumn + 35

Where above, if somecolumn uses this type, this method will be called with the value operator.add and 35. The return value is whatever SQLAlchemy type should be used for 35 for this particular operation.

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

Specify those Python types which should be coerced at the expression level to “IS <constant>” when compared using == (and same for IS NOT in conjunction with !=).

For most SQLAlchemy types, this includes NoneType, as well as bool.

TypeDecorator modifies this list to only include NoneType, as typedecorator implementations that deal with boolean types are common.

Custom TypeDecorator classes can override this attribute to return an empty tuple, in which case no values will be coerced to constants.

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

Given a SELECT column expression, return a wrapping SQL expression.

备注

This method is called during the SQL compilation phase of a statement, when rendering a SQL string. It is not called against specific values, and should not be confused with the TypeDecorator.process_result_value() method, which is the more typical method that processes the actual value returned in a result row subsequent to statement execution time.

Subclasses of TypeDecorator can override this method to provide custom column expression behavior for the type. This implementation will replace that of the underlying implementation type.

See the description of TypeEngine.column_expression() for a complete description of the method’s use.

attribute sqlalchemy.types.TypeDecorator.comparator_factory: _ComparatorFactory[Any]¶

A Comparator class which will apply to operations performed by owning ColumnElement objects.

The comparator_factory attribute is a hook consulted by the core expression system when column and SQL expression operations are performed. When a Comparator class is associated with this attribute, it allows custom re-definition of all existing operators, as well as definition of new operators. Existing operators include those provided by Python operator overloading such as ColumnOperators.__add__() and ColumnOperators.__eq__(), those provided as standard attributes of ColumnOperators such as ColumnOperators.like() and ColumnOperators.in_().

Rudimentary usage of this hook is allowed through simple subclassing of existing types, or alternatively by using TypeDecorator. See the documentation section 重新定义和创建新运算符 for examples.

method sqlalchemy.types.TypeDecorator.compare_values(x: Any, y: Any) → bool¶

Given two values, compare them for equality.

By default this calls upon TypeEngine.compare_values() of the underlying “impl”, which in turn usually uses the Python equals operator ==.

This function is used by the ORM to compare an original-loaded value with an intercepted “changed” value, to determine if a net change has occurred.

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

Produce a copy of this TypeDecorator instance.

This is a shallow copy and is provided to fulfill part of the TypeEngine contract. It usually does not need to be overridden unless the user-defined TypeDecorator has local state that should be deep-copied.

method sqlalchemy.types.TypeDecorator.get_dbapi_type(dbapi: module) → Any | None¶

Return the DBAPI type object represented by this TypeDecorator.

By default this calls upon TypeEngine.get_dbapi_type() of the underlying “impl”.

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

Provide a literal processing function for the given Dialect.

This is the method that fulfills the TypeEngine contract for literal value conversion which normally occurs via the TypeEngine.literal_processor() method.

备注

User-defined subclasses of TypeDecorator should not implement this method, and should instead implement TypeDecorator.process_literal_param() so that the “inner” processing provided by the implementing type is maintained.

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

Return a TypeEngine object corresponding to a dialect.

This is an end-user override hook that can be used to provide differing types depending on the given dialect. It is used by the TypeDecorator implementation of type_engine() to help determine what type should ultimately be returned for a given TypeDecorator.

By default returns self.impl.

method sqlalchemy.types.TypeDecorator.process_bind_param(value: _T | None, dialect: Dialect) → Any¶

Receive a bound parameter value to be converted.

Custom subclasses of TypeDecorator should override this method to provide custom behaviors for incoming data values. This method is called at statement execution time and is passed the literal Python data value which is to be associated with a bound parameter in the statement.

The operation could be anything desired to perform custom behavior, such as transforming or serializing data. This could also be used as a hook for validating logic.

参数:

value¶ – Data to operate upon, of any type expected by this method in the subclass. Can be None.
dialect¶ – the Dialect in use.

参见

增强现有类型

TypeDecorator.process_result_value()

method sqlalchemy.types.TypeDecorator.process_literal_param(value: _T | None, dialect: Dialect) → str¶

Receive a literal parameter value to be rendered inline within a statement.

备注

This method is called during the SQL compilation phase of a statement, when rendering a SQL string. Unlike other SQL compilation methods, it is passed a specific Python value to be rendered as a string. However it should not be confused with the TypeDecorator.process_bind_param() method, which is the more typical method that processes the actual value passed to a particular parameter at statement execution time.

Custom subclasses of TypeDecorator should override this method to provide custom behaviors for incoming data values that are in the special case of being rendered as literals.

The returned string will be rendered into the output string.

method sqlalchemy.types.TypeDecorator.process_result_value(value: Any | None, dialect: Dialect) → _T | None¶

Receive a result-row column value to be converted.

Custom subclasses of TypeDecorator should override this method to provide custom behaviors for data values being received in result rows coming from the database. This method is called at result fetching time and is passed the literal Python data value that’s extracted from a database result row.

The operation could be anything desired to perform custom behavior, such as transforming or deserializing data.

参数:

value¶ – Data to operate upon, of any type expected by this method in the subclass. Can be None.
dialect¶ – the Dialect in use.

参见

增强现有类型

TypeDecorator.process_bind_param()

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

Provide a result value processing function for the given Dialect.

This is the method that fulfills the TypeEngine contract for bound value conversion which normally occurs via the TypeEngine.result_processor() method.

备注

User-defined subclasses of TypeDecorator should not implement this method, and should instead implement TypeDecorator.process_result_value() so that the “inner” processing provided by the implementing type is maintained.

参数:

dialect¶ – Dialect instance in use.
coltype¶ – A SQLAlchemy data type

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

A sorting function that can be passed as the key to sorted.

The default value of None indicates that the values stored by this type are self-sorting.

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

Return a dialect-specific TypeEngine instance for this TypeDecorator.

In most cases this returns a dialect-adapted form of the TypeEngine type represented by self.impl. Makes usage of dialect_impl(). Behavior can be customized here by overriding load_dialect_impl().

英文

The TypeDecorator allows the creation of custom types which add bind-parameter and result-processing behavior to an existing type object. It is used when additional in-Python marshalling of data to and/or from the database is required.

The bind- and result-processing of TypeDecorator is in addition to the processing already performed by the hosted type, which is customized by SQLAlchemy on a per-DBAPI basis to perform processing specific to that DBAPI. While it is possible to replace this handling for a given type through direct subclassing, it is never needed in practice and SQLAlchemy no longer supports this as a public use case.

The TypeDecorator can be used to provide a consistent means of converting some type of value as it is passed into and out of the database. When using the ORM, a similar technique exists for converting user data from arbitrary formats which is to use the validates() decorator. This technique may be more appropriate when data coming into an ORM model needs to be normalized in some way that is specific to the business case and isn’t as generic as a datatype.

class sqlalchemy.types.TypeDecorator¶

Allows the creation of types which add additional functionality to an existing type.

This method is preferred to direct subclassing of SQLAlchemy’s built-in types as it ensures that all required functionality of the underlying type is kept in place.

Typical usage:

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)

Our expression of somecol + date with the above type will coerce the “date” on the right side to also be treated as MyEpochType.

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)

Without the above step, index operations such as mycol['foo'] will cause the index value 'foo' to be JSON encoded.

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)

Class signature

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

class Comparator¶

A Comparator that is specific to TypeDecorator.

User-defined TypeDecorator classes should not typically need to modify this.

Class signature

class sqlalchemy.types.TypeDecorator.Comparator (sqlalchemy.types.Comparator)

method sqlalchemy.types.TypeDecorator.Comparator.operate(op: OperatorType, *other: Any, **kwargs: Any) → ColumnElement[_CT]¶

Operate on an argument.

This is the lowest level of operation, raises NotImplementedError by default.

Overriding this on a subclass can allow common behavior to be applied to all operations. For example, overriding ColumnOperators to apply func.lower() to the left and right side:

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

参数:

op¶ – Operator callable.
*other¶ – the ‘other’ side of the operation. Will be a single scalar for most operations.
**kwargs¶ – modifiers. These may be passed by special operators such as ColumnOperators.contains().

method sqlalchemy.types.TypeDecorator.Comparator.reverse_operate(op: OperatorType, other: Any, **kwargs: Any) → ColumnElement[_CT]¶

Reverse operate on an argument.

Usage is the same as operate().

method sqlalchemy.types.TypeDecorator.__init__(*args: Any, **kwargs: Any)¶

Construct a TypeDecorator.

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute ‘as-is’, ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

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

Given a bind value (i.e. a BindParameter instance), return a SQL expression which will typically wrap the given parameter.

备注

Subclasses of TypeDecorator can override this method to provide custom bind expression behavior for the type. This implementation will replace that of the underlying implementation type.

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

Provide a bound value processing function for the given Dialect.

This is the method that fulfills the TypeEngine contract for bound value conversion which normally occurs via the TypeEngine.bind_processor() method.

备注

参数:: dialect¶ – Dialect instance in use.

method sqlalchemy.types.TypeDecorator.coerce_compared_value(op: OperatorType | None, value: Any) → Any¶

Suggest a type for a ‘coerced’ Python value in an expression.

expr = table.c.somecolumn + 35

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

Specify those Python types which should be coerced at the expression level to “IS <constant>” when compared using == (and same for IS NOT in conjunction with !=).

For most SQLAlchemy types, this includes NoneType, as well as bool.

TypeDecorator modifies this list to only include NoneType, as typedecorator implementations that deal with boolean types are common.

Custom TypeDecorator classes can override this attribute to return an empty tuple, in which case no values will be coerced to constants.

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

Given a SELECT column expression, return a wrapping SQL expression.

备注

Subclasses of TypeDecorator can override this method to provide custom column expression behavior for the type. This implementation will replace that of the underlying implementation type.

See the description of TypeEngine.column_expression() for a complete description of the method’s use.

attribute sqlalchemy.types.TypeDecorator.comparator_factory: _ComparatorFactory[Any]¶

A Comparator class which will apply to operations performed by owning ColumnElement objects.

method sqlalchemy.types.TypeDecorator.compare_values(x: Any, y: Any) → bool¶

Given two values, compare them for equality.

By default this calls upon TypeEngine.compare_values() of the underlying “impl”, which in turn usually uses the Python equals operator ==.

This function is used by the ORM to compare an original-loaded value with an intercepted “changed” value, to determine if a net change has occurred.

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

Produce a copy of this TypeDecorator instance.

method sqlalchemy.types.TypeDecorator.get_dbapi_type(dbapi: module) → Any | None¶

Return the DBAPI type object represented by this TypeDecorator.

By default this calls upon TypeEngine.get_dbapi_type() of the underlying “impl”.

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

Provide a literal processing function for the given Dialect.

This is the method that fulfills the TypeEngine contract for literal value conversion which normally occurs via the TypeEngine.literal_processor() method.

备注

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

Return a TypeEngine object corresponding to a dialect.

By default returns self.impl.

method sqlalchemy.types.TypeDecorator.process_bind_param(value: _T | None, dialect: Dialect) → Any¶

Receive a bound parameter value to be converted.

The operation could be anything desired to perform custom behavior, such as transforming or serializing data. This could also be used as a hook for validating logic.

参数:

value¶ – Data to operate upon, of any type expected by this method in the subclass. Can be None.
dialect¶ – the Dialect in use.

参见

增强现有类型

TypeDecorator.process_result_value()

method sqlalchemy.types.TypeDecorator.process_literal_param(value: _T | None, dialect: Dialect) → str¶

Receive a literal parameter value to be rendered inline within a statement.

备注

Custom subclasses of TypeDecorator should override this method to provide custom behaviors for incoming data values that are in the special case of being rendered as literals.

The returned string will be rendered into the output string.

method sqlalchemy.types.TypeDecorator.process_result_value(value: Any | None, dialect: Dialect) → _T | None¶

Receive a result-row column value to be converted.

The operation could be anything desired to perform custom behavior, such as transforming or deserializing data.

参数:

value¶ – Data to operate upon, of any type expected by this method in the subclass. Can be None.
dialect¶ – the Dialect in use.

参见

增强现有类型

TypeDecorator.process_bind_param()

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

Provide a result value processing function for the given Dialect.

This is the method that fulfills the TypeEngine contract for bound value conversion which normally occurs via the TypeEngine.result_processor() method.

备注

参数:

dialect¶ – Dialect instance in use.
coltype¶ – A SQLAlchemy data type

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

A sorting function that can be passed as the key to sorted.

The default value of None indicates that the values stored by this type are self-sorting.

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

Return a dialect-specific TypeEngine instance for this TypeDecorator.

应用 SQL 级绑定/结果处理¶

Applying SQL-level Bind/Result Processing

中文

如在增强现有类型小节中所示， 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 应用于列子句中，使返回值在进入结果集前经过函数处理，而 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 表达式不会干扰包装表达式的标签。例如，当我们对表达式使用 label() 并执行 select() 时，标签会被渲染在最外层:

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):
        # 将绑定值从 PGPString 转换为 String，
        # 以便原样传递给 psycopg2，而不会被 dbapi.Binary 包裹
        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'}

英文

As seen in the section 增强现有类型, SQLAlchemy allows Python functions to be invoked both when parameters are sent to a statement, as well as when result rows are loaded from the database, to apply transformations to the values as they are sent to or from the database. It is also possible to define SQL-level transformations as well. The rationale here is when only the relational database contains a particular series of functions that are necessary to coerce incoming and outgoing data between an application and persistence format. Examples include using database-defined encryption/decryption functions, as well as stored procedures that handle geographic data.

Any TypeEngine, UserDefinedType or TypeDecorator subclass can include implementations of TypeEngine.bind_expression() and/or TypeEngine.column_expression(), which when defined to return a non-None value should return a ColumnElement expression to be injected into the SQL statement, either surrounding bound parameters or a column expression. For example, to build a Geometry type which will apply the PostGIS function ST_GeomFromText to all outgoing values and the function ST_AsText to all incoming data, we can create our own subclass of UserDefinedType which provides these methods in conjunction with 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)

We can apply the Geometry type into Table metadata and use it in a select() construct:

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

The resulting SQL embeds both functions as appropriate. ST_AsText is applied to the columns clause so that the return value is run through the function before passing into a result set, and ST_GeomFromText is run on the bound parameter so that the passed-in value is converted:

SELECT geometry.geom_id, ST_AsText(geometry.geom_data) AS geom_data_1
FROM geometry
WHERE geometry.geom_data = ST_GeomFromText(:geom_data_2)

The TypeEngine.column_expression() method interacts with the mechanics of the compiler such that the SQL expression does not interfere with the labeling of the wrapped expression. Such as, if we rendered a select() against a label() of our expression, the string label is moved to the outside of the wrapped expression:

print(select(geometry.c.geom_data.label("my_data")))

Output:

SELECT ST_AsText(geometry.geom_data) AS my_data
FROM geometry

Another example is we decorate BYTEA to provide a PGPString, which will make use of the PostgreSQL pgcrypto extension to encrypt/decrypt values transparently:

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

The pgp_sym_encrypt and pgp_sym_decrypt functions are applied to the INSERT and SELECT statements:

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

重新定义和创建新运算符¶

Redefining and Creating New Operators

中文

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 操作符，而上述方法并未直接支持时，最简便的方式是使用任何 SQL 表达式对象的 Operators.op() 方法；该方法接受一个表示 SQL 操作符的字符串，并返回一个可调用对象（Python callable），该对象接受任意右侧表达式作为参数：

>>> from sqlalchemy importcolumn>>> expr=column("x").op(">>")(column("y"))>>> print(expr)x >> y

当使用自定义 SQL 类型时，也可以像上述一样实现自定义操作符，使其自动出现在使用该列类型的任何列表达式上，无需每次手动调用 Operators.op()。

为实现此功能，SQL 表达式构造会查找与之关联的 TypeEngine 对象，以确定内建操作符的行为，并查找是否有自定义方法被调用。 TypeEngine 定义了一个由 Comparator 类实现的 “comparison” 对象，用于提供 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 属性设定为一个新的类，该类继承自与 Integer 类型相关的 Comparator 类。

用法示例：

>>> sometable=Table("sometable",metadata,Column("data",MyInt))>>> print(sometable.c.data+5)sometable.data goofy :data_1

ColumnOperators.__add__() 的实现会被 SQL 表达式本体调用，它通过将自身作为 expr 属性来实例化 Comparator。该属性可在实现中用于直接引用原始的 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 表达式对象上，从而使这些方法成为 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 importcolumn>>> print(column("x",MyInteger).factorial())x !

参见

Operators.op()

TypeEngine.comparator_factory

英文

SQLAlchemy Core defines a fixed set of expression operators available to all column expressions. Some of these operations have the effect of overloading Python’s built-in operators; examples of such operators include ColumnOperators.__eq__() (table.c.somecolumn == 'foo'), ColumnOperators.__invert__() (~table.c.flag), and ColumnOperators.__add__() (table.c.x + table.c.y). Other operators are exposed as explicit methods on column expressions, such as ColumnOperators.in_() (table.c.value.in_(['x', 'y'])) and ColumnOperators.like() (table.c.value.like('%ed%')).

When the need arises for a SQL operator that isn’t directly supported by the already supplied methods above, the most expedient way to produce this operator is to use the Operators.op() method on any SQL expression object; this method is given a string representing the SQL operator to render, and the return value is a Python callable that accepts any arbitrary right-hand side expression:

>>> from sqlalchemy importcolumn>>> expr=column("x").op(">>")(column("y"))>>> print(expr)x >> y

When making use of custom SQL types, there is also a means of implementing custom operators as above that are automatically present upon any column expression that makes use of that column type, without the need to directly call Operators.op() each time the operator is to be used.

To achieve this, a SQL expression construct consults the TypeEngine object associated with the construct in order to determine the behavior of the built-in operators as well as to look for new methods that may have been invoked. TypeEngine defines a “comparison” object implemented by the Comparator class to provide the base behavior for SQL operators, and many specific types provide their own sub-implementations of this class. User-defined Comparator implementations can be built directly into a simple subclass of a particular type in order to override or define new operations. Below, we create a Integer subclass which overrides the ColumnOperators.__add__() operator, which in turn uses Operators.op() to produce the custom SQL itself:

from sqlalchemy import Integer


class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def __add__(self, other):
            return self.op("goofy")(other)

The above configuration creates a new class MyInt, which establishes the TypeEngine.comparator_factory attribute as referring to a new class, subclassing the Comparator class associated with the Integer type.

Usage:

>>> sometable=Table("sometable",metadata,Column("data",MyInt))>>> print(sometable.c.data+5)sometable.data goofy :data_1

The implementation for ColumnOperators.__add__() is consulted by an owning SQL expression, by instantiating the Comparator with itself as the expr attribute. This attribute may be used when the implementation needs to refer to the originating ColumnElement object directly:

from sqlalchemy import Integer


class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def __add__(self, other):
            return func.special_addition(self.expr, other)

New methods added to a Comparator are exposed on an owning SQL expression object using a dynamic lookup scheme, which exposes methods added to Comparator onto the owning ColumnElement expression construct. For example, to add a log() function to integers:

from sqlalchemy import Integer, func


class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def log(self, other):
            return func.log(self.expr, other)

Using the above type:

>>> print(sometable.c.data.log(5))log(:log_1, :log_2)

When using Operators.op() for comparison operations that return a boolean result, the Operators.op.is_comparison flag should be set to True:

class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def is_frobnozzled(self, other):
            return self.op("--is_frobnozzled->", is_comparison=True)(other)

Unary operations are also possible. For example, to add an implementation of the PostgreSQL factorial operator, we combine the UnaryExpression construct along with a custom_op to produce the factorial expression:

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
            )

Using the above type:

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

参见

Operators.op()

TypeEngine.comparator_factory

创建新类型¶

Creating New Types

中文

UserDefinedType 类是一个简单的基类，用于定义全新的数据库类型。使用它来表示 SQLAlchemy 不支持的原生数据库类型。如果只需要 Python 翻译行为，请使用 TypeDecorator。

class sqlalchemy.types.UserDefinedType¶

Base for user defined types.

This should be the base of new types. Note that for most cases, TypeDecorator is probably more appropriate:

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

Once the type is made, it’s immediately usable:

table = Table(
    "foo",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("data", MyType(16)),
)

The get_col_spec() method will in most cases receive a keyword argument type_expression which refers to the owning expression of the type as being compiled, such as a Column or cast() construct. This keyword is only sent if the method accepts keyword arguments (e.g. **kw) in its argument signature; introspection is used to check for this in order to support legacy forms of this function.

The UserDefinedType.cache_ok class-level flag indicates if this custom UserDefinedType is safe to be used as part of a cache key. This flag defaults to None which will initially generate a warning when the SQL compiler attempts to generate a cache key for a statement that uses this type. If the UserDefinedType is not guaranteed to produce the same bind/result behavior and SQL generation every time, this flag should be set to False; otherwise if the class produces the same behavior each time, it may be set to True. See UserDefinedType.cache_ok for further notes on how this works.

在 1.4.28 版本加入: Generalized the ExternalType.cache_ok flag so that it is available for both TypeDecorator as well as UserDefinedType.

Class signature

class sqlalchemy.types.UserDefinedType (sqlalchemy.types.ExternalType, sqlalchemy.types.TypeEngineMixin, sqlalchemy.types.TypeEngine, sqlalchemy.util.langhelpers.EnsureKWArg)

attribute sqlalchemy.types.UserDefinedType.cache_ok: bool | None = None¶

inherited from the ExternalType.cache_ok attribute of ExternalType

Indicate if statements using this ExternalType are “safe to cache”.

The default value None will emit a warning and then not allow caching of a statement which includes this type. Set to False to disable statements using this type from being cached at all without a warning. When set to True, the object’s class and selected elements from its state will be used as part of the cache key. For example, using a TypeDecorator:

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

    def __init__(self, choices):
        self.choices = tuple(choices)
        self.internal_only = True

The cache key for the above type would be equivalent to:

>>> MyType(["a", "b", "c"])._static_cache_key
(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

The caching scheme will extract attributes from the type that correspond to the names of parameters in the __init__() method. Above, the “choices” attribute becomes part of the cache key but “internal_only” does not, because there is no parameter named “internal_only”.

The requirements for cacheable elements is that they are hashable and also that they indicate the same SQL rendered for expressions using this type every time for a given cache value.

To accommodate for datatypes that refer to unhashable structures such as dictionaries, sets and lists, these objects can be made “cacheable” by assigning hashable structures to the attributes whose names correspond with the names of the arguments. For example, a datatype which accepts a dictionary of lookup values may publish this as a sorted series of tuples. Given a previously un-cacheable type as:

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

Where “lookup” is a dictionary. The type will not be able to generate a cache key:

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

If we did set up such a cache key, it wouldn’t be usable. We would get a tuple structure that contains a dictionary inside of it, which cannot itself be used as a key in a “cache dictionary” such as SQLAlchemy’s statement cache, since Python dictionaries aren’t hashable:

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

The type may be made cacheable by assigning a sorted tuple of tuples to the “.lookup” attribute:

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

Where above, the cache key for LookupType({"a": 10, "b": 20}) will be:

>>> 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: OperatorType | None, value: Any) → TypeEngine[Any]¶

Suggest a type for a ‘coerced’ Python value in an expression.

Default behavior for UserDefinedType is the same as that of TypeDecorator; by default it returns self, assuming the compared value should be coerced into the same type as this one. See TypeDecorator.coerce_compared_value() for more detail.

attribute sqlalchemy.types.UserDefinedType.ensure_kwarg: str = 'get_col_spec'¶

a regular expression that indicates method names for which the method should accept **kw arguments.

The class will scan for methods matching the name template and decorate them if necessary to ensure **kw parameters are accepted.

英文

The UserDefinedType class is provided as a simple base class for defining entirely new database types. Use this to represent native database types not known by SQLAlchemy. If only Python translation behavior is needed, use TypeDecorator instead.

class sqlalchemy.types.UserDefinedType¶

Base for user defined types.

This should be the base of new types. Note that for most cases, TypeDecorator is probably more appropriate:

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

Once the type is made, it’s immediately usable:

table = Table(
    "foo",
    metadata_obj,
    Column("id", Integer, primary_key=True),
    Column("data", MyType(16)),
)

在 1.4.28 版本加入: Generalized the ExternalType.cache_ok flag so that it is available for both TypeDecorator as well as UserDefinedType.

Class signature

class sqlalchemy.types.UserDefinedType (sqlalchemy.types.ExternalType, sqlalchemy.types.TypeEngineMixin, sqlalchemy.types.TypeEngine, sqlalchemy.util.langhelpers.EnsureKWArg)

attribute sqlalchemy.types.UserDefinedType.cache_ok: bool | None = None¶

inherited from the ExternalType.cache_ok attribute of ExternalType

Indicate if statements using this ExternalType are “safe to cache”.

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

    def __init__(self, choices):
        self.choices = tuple(choices)
        self.internal_only = True

The cache key for the above type would be equivalent to:

>>> MyType(["a", "b", "c"])._static_cache_key
(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

The requirements for cacheable elements is that they are hashable and also that they indicate the same SQL rendered for expressions using this type every time for a given cache value.

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

Where “lookup” is a dictionary. The type will not be able to generate a cache key:

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

The type may be made cacheable by assigning a sorted tuple of tuples to the “.lookup” attribute:

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

Where above, the cache key for LookupType({"a": 10, "b": 20}) will be:

>>> 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: OperatorType | None, value: Any) → TypeEngine[Any]¶

Suggest a type for a ‘coerced’ Python value in an expression.

attribute sqlalchemy.types.UserDefinedType.ensure_kwarg: str = 'get_col_spec'¶

a regular expression that indicates method names for which the method should accept **kw arguments.

The class will scan for methods matching the name template and decorate them if necessary to ensure **kw parameters are accepted.

使用自定义类型和反射¶

Working with Custom Types and Reflection

中文

需要注意的是，那些被修改以在 Python 中具有附加行为的数据库类型（包括基于 TypeDecorator 的类型以及其他用户定义的数据类型子类），在数据库架构中并没有任何表示形式。当使用反射数据库对象中描述的数据库自省功能时， SQLAlchemy 使用的是一组固定的映射，用于将数据库服务器返回的数据类型信息关联到 SQLAlchemy 的数据类型对象上。例如，当我们查看 PostgreSQL 架构中某个数据库列的定义时，可能会得到字符串 "VARCHAR"。 SQLAlchemy 的 PostgreSQL 方言包含一个硬编码的映射，它将字符串 "VARCHAR" 关联到 SQLAlchemy 的 VARCHAR 类。因此，当我们执行类似 Table('my_table', m, autoload_with=engine) 的语句时，其中的 Column 对象会包含一个 VARCHAR 实例。

这意味着，如果一个 Table 对象使用了与数据库原生类型名称不直接对应的类型对象，那么当我们在其他地方针对该数据库表、并通过反射创建一个新的 Table 对象时，它将不会包含该自定义类型。例如：

>>> 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，它是一个基于 LargeBinary 类型的 TypeDecorator，而在 SQLite 中该类型对应于数据库类型 BLOB 。在 CREATE TABLE 语句中，我们看到使用的是 BLOB 类型。 SQLite 数据库对我们所用的 PickleType 一无所知。

如果我们查看 my_table.c.data.type 的数据类型，由于这是我们直接创建的 Python 对象，它的类型是 PickleType:

>>> my_table.c.data.type
PickleType()

然而，如果我们使用反射来创建另一个 Table 实例，由于我们创建的 SQLite 数据库中并未保存 PickleType 的信息，因此得到的是 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 元数据已经存在。然而，在某些情况下，应用程序（或多个应用程序）需要同时使用包含自定义 Python 类型的显式 Table 元数据和通过数据库反射生成的 Table 对象，并希望后者仍然具备自定义数据类型的 Python 行为，此时就需要采取额外步骤。

最直接的方式是覆盖某些列，如覆盖反射列中所述。这种技术结合了反射和显式的 Column 对象，用于那些需要使用自定义或装饰类型的列:

>>> metadata_three = MetaData()
>>> my_reflected_table = Table(
...     "my_table",
...     metadata_three,
...     Column("data", PickleType),
...     autoload_with=engine,
... )

上述 my_reflected_table 对象是通过反射加载的，它会从 SQLite 数据库中加载 “id” 列的定义。但对于 “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()

当上述代码在任何表反射发生之前被调用时（注意它应该在应用程序中 只调用一次，因为它是全局规则），对于任何包含 BLOB 数据类型的 Table，反射过程中得到的类型将会在 Column 对象中被存储为 PickleType。

在实际使用中，上述基于事件的方法通常会包含额外的规则，以仅影响那些确实需要转换数据类型的列，例如使用一个包含表名或列名的查找表，或使用其他启发式逻辑，以更准确地确定哪些列应该设定为 Python 数据类型。

英文

It is important to note that database types which are modified to have additional in-Python behaviors, including types based on TypeDecorator as well as other user-defined subclasses of datatypes, do not have any representation within a database schema. When using database the introspection features described at 反射数据库对象, SQLAlchemy makes use of a fixed mapping which links the datatype information reported by a database server to a SQLAlchemy datatype object. For example, if we look inside of a PostgreSQL schema at the definition for a particular database column, we might receive back the string "VARCHAR". SQLAlchemy’s PostgreSQL dialect has a hardcoded mapping which links the string name "VARCHAR" to the SQLAlchemy VARCHAR class, and that’s how when we emit a statement like Table('my_table', m, autoload_with=engine), the Column object within it would have an instance of VARCHAR present inside of it.

The implication of this is that if a Table object makes use of type objects that don’t correspond directly to the database-native type name, if we create a new Table object against a new MetaData collection for this database table elsewhere using reflection, it will not have this datatype. For example:

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

Above, we made use of PickleType, which is a TypeDecorator that works on top of the LargeBinary datatype, which on SQLite corresponds to the database type BLOB. In the CREATE TABLE, we see that the BLOB datatype is used. The SQLite database knows nothing about the PickleType we’ve used.

If we look at the datatype of my_table.c.data.type, as this is a Python object that was created by us directly, it is PickleType:

>>> my_table.c.data.type
PickleType()

However, if we create another instance of Table using reflection, the use of PickleType is not represented in the SQLite database we’ve created; we instead get back 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()

Typically, when an application defines explicit Table metadata with custom types, there is no need to use table reflection because the necessary Table metadata is already present. However, for the case where an application, or a combination of them, need to make use of both explicit Table metadata which includes custom, Python-level datatypes, as well as Table objects which set up their Column objects as reflected from the database, which nevertheless still need to exhibit the additional Python behaviors of the custom datatypes, additional steps must be taken to allow this.

The most straightforward is to override specific columns as described at 覆盖反射列. In this technique, we simply use reflection in combination with explicit Column objects for those columns for which we want to use a custom or decorated datatype:

>>> metadata_three = MetaData()
>>> my_reflected_table = Table(
...     "my_table",
...     metadata_three,
...     Column("data", PickleType),
...     autoload_with=engine,
... )

The my_reflected_table object above is reflected, and will load the definition of the “id” column from the SQLite database. But for the “data” column, we’ve overridden the reflected object with an explicit Column definition that includes our desired in-Python datatype, the PickleType. The reflection process will leave this Column object intact:

>>> my_reflected_table.c.data.type
PickleType()

A more elaborate way to convert from database-native type objects to custom datatypes is to use the DDLEvents.column_reflect() event handler. If for example we knew that we wanted all BLOB datatypes to in fact be PickleType, we could set up a rule across the board:

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

When the above code is invoked before any table reflection occurs (note also it should be invoked only once in the application, as it is a global rule), upon reflecting any Table that includes a column with a BLOB datatype, the resulting datatype will be stored in the Column object as PickleType.

In practice, the above event-based approach would likely have additional rules in order to affect only those columns where the datatype is important, such as a lookup table of table names and possibly column names, or other heuristics in order to accurately determine which columns should be established with an in Python datatype.

自定义类型¶

重写类型编译¶

增强现有类型¶

TypeDecorator 配方¶

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

四舍五入数字¶

将时区感知时间戳存储为时区简单 UTC¶

后端无关的 GUID 类型¶

将 Python `uuid.UUID` 链接到自定义类型以进行 ORM 映射¶

编组 JSON 字符串¶

添加可变性¶

处理比较操作¶

应用 SQL 级绑定/结果处理¶

重新定义和创建新运算符¶

创建新类型¶

使用自定义类型和反射¶

自定义类型¶

重写类型编译¶

增强现有类型¶

TypeDecorator 配方¶

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

四舍五入数字¶

将时区感知时间戳存储为时区简单 UTC¶

后端无关的 GUID 类型¶

将 Python uuid.UUID 链接到自定义类型以进行 ORM 映射¶

编组 JSON 字符串¶

添加可变性¶

处理比较操作¶

应用 SQL 级绑定/结果处理¶

重新定义和创建新运算符¶

创建新类型¶

使用自定义类型和反射¶

将 Python `uuid.UUID` 链接到自定义类型以进行 ORM 映射¶