Base64Bytes¶

class lsst.daf.butler.ddl.Base64Bytes(nbytes: int | None = None, *args: Any, **kwargs: Any)¶

Bases: TypeDecorator

A SQLAlchemy custom type for Python bytes.

Maps Python bytes to a base64-encoded sqlalchemy.Text field.

Attributes Summary

`cache_ok`	Indicate if statements using this `ExternalType` are "safe to cache".
`coerce_to_is_types`	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 `!=`).
`comparator_factory`	Base class for custom comparison operations defined at the type level.
`dispatch`
`hashable`	Flag, if False, means values from this type aren't hashable.
`impl_instance`
`python_type`	Return the Python type object expected to be returned by instances of this type, if known.
`render_bind_cast`	Render bind casts for `BindTyping.RENDER_CASTS` mode.
`render_literal_cast`	render casts when rendering a value as an inline literal, e.g.
`should_evaluate_none`	If True, the Python constant `None` is considered to be handled explicitly by this type.
`sort_key_function`

Methods Summary

`adapt`(cls, **kw)	Produce an "adapted" form of this type, given an "impl" class to work with.
`as_generic`([allow_nulltype])	Return an instance of the generic type corresponding to this type using heuristic rule.
`bind_expression`(bindparam)	Given a bind value (i.e.
`bind_processor`(dialect)	Provide a bound value processing function for the given `Dialect`.
`coerce_compared_value`(op, value)	Suggest a type for a 'coerced' Python value in an expression.
`column_expression`(column)	Given a SELECT column expression, return a wrapping SQL expression.
`compare_against_backend`(dialect, conn_type)	Compare this type against the given backend type.
`compare_values`(x, y)	Given two values, compare them for equality.
`compile`([dialect])	Produce a string-compiled form of this `TypeEngine`.
`copy`(**kw)	Produce a copy of this `TypeDecorator` instance.
`copy_value`(value)
`dialect_impl`(dialect)	Return a dialect-specific implementation for this `TypeEngine`.
`evaluates_none`()	Return a copy of this type which has the `should_evaluate_none` flag set to True.
`get_dbapi_type`(dbapi)	Return the DBAPI type object represented by this `TypeDecorator`.
`literal_processor`(dialect)	Provide a literal processing function for the given `Dialect`.
`load_dialect_impl`(dialect)	Return a `TypeEngine` object corresponding to a dialect.
`process_bind_param`(value, dialect)	Receive a bound parameter value to be converted.
`process_literal_param`(value, dialect)	Receive a literal parameter value to be rendered inline within a statement.
`process_result_value`(value, dialect)	Receive a result-row column value to be converted.
`result_processor`(dialect, coltype)	Provide a result value processing function for the given `Dialect`.
`type_engine`(dialect)	Return a dialect-specific `TypeEngine` instance for this `TypeDecorator`.
`with_variant`(type_, *dialect_names)	Produce a copy of this type object that will utilize the given type when applied to the dialect of the given name.

Attributes Documentation

cache_ok: bool | None = True¶

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

New in version 1.4.14: - added the cache_ok flag to allow some configurability of caching for TypeDecorator classes.

New in version 1.4.28: - added the ExternalType mixin which generalizes the cache_ok flag to both the TypeDecorator and UserDefinedType classes.

See also

sql_caching

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.

comparator_factory: _ComparatorFactory[Any]¶

A TypeEngine.Comparator class which will apply to operations performed by owning _expression.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 TypeEngine.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 operators.ColumnOperators.__add__() and operators.ColumnOperators.__eq__(), those provided as standard attributes of operators.ColumnOperators such as operators.ColumnOperators.like() and operators.ColumnOperators.in_().

Rudimentary usage of this hook is allowed through simple subclassing of existing types, or alternatively by using TypeDecorator. See the documentation section types_operators for examples.

dispatch: dispatcher[SchemaEventTarget] = <sqlalchemy.event.base.DDLEventsDispatch object>¶

hashable = True¶

Flag, if False, means values from this type aren’t hashable.

Used by the ORM when uniquing result lists.

impl_instance¶

python_type¶

render_bind_cast = False¶

Render bind casts for BindTyping.RENDER_CASTS mode.

If True, this type (usually a dialect level impl type) signals to the compiler that a cast should be rendered around a bound parameter for this type.

New in version 2.0.

See also

BindTyping

render_literal_cast = False¶: render casts when rendering a value as an inline literal, e.g. with TypeEngine.literal_processor().

New in version 2.0.

should_evaluate_none: bool = False¶

If True, the Python constant None is considered to be handled explicitly by this type.

The ORM uses this flag to indicate that a positive value of None is passed to the column in an INSERT statement, rather than omitting the column from the INSERT statement which has the effect of firing off column-level defaults. It also allows types which have special behavior for Python None, such as a JSON type, to indicate that they’d like to handle the None value explicitly.

To set this flag on an existing type, use the TypeEngine.evaluates_none() method.

See also

TypeEngine.evaluates_none()

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.

New in version 1.3.8.

Methods Documentation

adapt(cls: Type[TypeEngine | TypeEngineMixin], **kw: Any) → TypeEngine¶

Produce an “adapted” form of this type, given an “impl” class to work with.

This method is used internally to associate generic types with “implementation” types that are specific to a particular dialect.

as_generic(allow_nulltype: bool = False) → TypeEngine¶

Return an instance of the generic type corresponding to this type using heuristic rule. The method may be overridden if this heuristic rule is not sufficient.

>>> from sqlalchemy.dialects.mysql import INTEGER
>>> INTEGER(display_width=4).as_generic()
Integer()

>>> from sqlalchemy.dialects.mysql import NVARCHAR
>>> NVARCHAR(length=100).as_generic()
Unicode(length=100)

New in version 1.4.0b2.

See also

metadata_reflection_dbagnostic_types - describes the use of _types.TypeEngine.as_generic() in conjunction with the _sql.DDLEvents.column_reflect() event, which is its intended use.

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.

Note

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 _types.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 _types.TypeDecorator can override this method to provide custom bind expression behavior for the type. This implementation will replace that of the underlying implementation type.

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 _types.TypeEngine.bind_processor() method.

Note

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

Parameters:: dialect – Dialect instance in use.

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.

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

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

Note

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 _types.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 _types.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 _types.TypeEngine.column_expression() for a complete description of the method’s use.

compare_against_backend(dialect: Dialect, conn_type: TypeEngine[Any]) → bool | None¶

Compare this type against the given backend type.

This function is currently not implemented for SQLAlchemy types, and for all built in types will return None. However, it can be implemented by a user-defined type where it can be consumed by schema comparison tools such as Alembic autogenerate.

A future release of SQLAlchemy will potentially implement this method for builtin types as well.

The function should return True if this type is equivalent to the given type; the type is typically reflected from the database so should be database specific. The dialect in use is also passed. It can also return False to assert that the type is not equivalent.

Parameters:

dialect – a Dialect that is involved in the comparison.
conn_type – the type object reflected from the backend.

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.

compile(dialect: Dialect | None = None) → str¶

Produce a string-compiled form of this TypeEngine.

When called with no arguments, uses a “default” dialect to produce a string result.

Parameters:: dialect – a Dialect instance.

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.

copy_value(value: Any) → Any¶

dialect_impl(dialect: Dialect) → TypeEngine[_T]¶: Return a dialect-specific implementation for this TypeEngine.

evaluates_none() → Self¶

Return a copy of this type which has the should_evaluate_none flag set to True.

E.g.:

Table(
    'some_table', metadata,
    Column(
        String(50).evaluates_none(),
        nullable=True,
        server_default='no value')
)

The ORM uses this flag to indicate that a positive value of None is passed to the column in an INSERT statement, rather than omitting the column from the INSERT statement which has the effect of firing off column-level defaults. It also allows for types which have special behavior associated with the Python None value to indicate that the value doesn’t necessarily translate into SQL NULL; a prime example of this is a JSON type which may wish to persist the JSON value 'null'.

In all cases, the actual NULL SQL value can be always be persisted in any column by using the _expression.null SQL construct in an INSERT statement or associated with an ORM-mapped attribute.

Note

The “evaluates none” flag does not apply to a value of None passed to :paramref:`_schema.Column.default` or :paramref:`_schema.Column.server_default`; in these cases, None still means “no default”.

See also

session_forcing_null - in the ORM documentation

:paramref:`.postgresql.JSON.none_as_null` - PostgreSQL JSON interaction with this flag.

TypeEngine.should_evaluate_none - class-level flag

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

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 _types.TypeEngine.literal_processor() method.

Note

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

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.

process_bind_param(value: bytes | None, dialect: Dialect) → str | None¶

Receive a bound parameter value to be converted.

Custom subclasses of _types.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.

Parameters:

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

See also

types_typedecorator

_types.TypeDecorator.process_result_value()

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

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

Note

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

process_result_value(value: str | None, dialect: Dialect) → bytes | None¶

Receive a result-row column value to be converted.

Custom subclasses of _types.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.

Parameters:

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

See also

types_typedecorator

_types.TypeDecorator.process_bind_param()

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 _types.TypeEngine.result_processor() method.

Note

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

Parameters:

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

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

with_variant(type_: _TypeEngineArgument[Any], *dialect_names: str) → Self¶

Produce a copy of this type object that will utilize the given type when applied to the dialect of the given name.

e.g.:

from sqlalchemy.types import String
from sqlalchemy.dialects import mysql

string_type = String()

string_type = string_type.with_variant(
    mysql.VARCHAR(collation='foo'), 'mysql', 'mariadb'
)

The variant mapping indicates that when this type is interpreted by a specific dialect, it will instead be transmuted into the given type, rather than using the primary type.

Changed in version 2.0: the _types.TypeEngine.with_variant() method now works with a _types.TypeEngine object “in place”, returning a copy of the original type rather than returning a wrapping object; the Variant class is no longer used.

Parameters:

type_ – a TypeEngine that will be selected as a variant from the originating type, when a dialect of the given name is in use.
*dialect_names –
one or more base names of the dialect which uses this type. (i.e. 'postgresql', 'mysql', etc.)

Changed in version 2.0: multiple dialect names can be specified for one variant.

See also

types_with_variant - illustrates the use of _types.TypeEngine.with_variant().

Navigation

Base64Bytes¶