types

SQLAlchemy custom types for use with the ORM.

class advanced_alchemy.types.GUID[source]

Bases: TypeDecorator

Platform-independent GUID type.

Uses PostgreSQL’s UUID type (Postgres, DuckDB, Cockroach), MSSQL’s UNIQUEIDENTIFIER type, Oracle’s RAW(16) type, otherwise uses BINARY(16) or CHAR(32), storing as stringified hex values.

Will accept stringified UUIDs as a hexstring or an actual UUID

impl = BINARY(length=16)
cache_ok = 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))))

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

Added 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 Compilation Caching

property python_type: type[UUID]

Return the Python type object expected to be returned by instances of this type, if known.

Basically, for those types which enforce a return type, or are known across the board to do such for all common DBAPIs (like int for example), will return that type.

If a return type is not defined, raises NotImplementedError.

Note that any type also accommodates NULL in SQL which means you can also get back None from any type in practice.

__init__(*args, binary=True, **kwargs)[source]

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.

Parameters:
Return type:

None

load_dialect_impl(dialect)[source]

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.

Return type:

Any

Parameters:

dialect (Dialect)

process_bind_param(value, dialect)[source]

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:
Return type:

bytes | str | None

See also

Augmenting Existing Types

_types.TypeDecorator.process_result_value()

Return type:

Union[bytes, str, None]

Parameters:
process_result_value(value, dialect)[source]

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:
Return type:

UUID | None

See also

Augmenting Existing Types

_types.TypeDecorator.process_bind_param()

Return type:

Optional[UUID]

Parameters:
static to_uuid(value)[source]
Return type:

Optional[UUID]

Parameters:

value (Any)

compare_values(x, y)[source]

Compare two values for equality.

Return type:

bool

Parameters:
class advanced_alchemy.types.ORA_JSONB[source]

Bases: TypeDecorator, SchemaType

Oracle Binary JSON type.

JsonB = _JSON().with_variant(PG_JSONB, “postgresql”).with_variant(ORA_JSONB, “oracle”)

impl

alias of BLOB

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

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

Added 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 Compilation Caching

property python_type: type[dict[str, Any]]

Return the Python type object expected to be returned by instances of this type, if known.

Basically, for those types which enforce a return type, or are known across the board to do such for all common DBAPIs (like int for example), will return that type.

If a return type is not defined, raises NotImplementedError.

Note that any type also accommodates NULL in SQL which means you can also get back None from any type in practice.

__init__(*args, **kwargs)[source]

Initialize JSON type

Parameters:
Return type:

None

coerce_compared_value(op, value)[source]

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.

Return type:

Any

Parameters:
load_dialect_impl(dialect)[source]

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.

Return type:

TypeEngine

Parameters:

dialect (Dialect)

process_bind_param(value, dialect)[source]

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 (Any) – Data to operate upon, of any type expected by this method in the subclass. Can be None.

  • dialect (Dialect) – the Dialect in use.

  • value (Any)

  • dialect (Dialect)

Return type:

Any | None

See also

Augmenting Existing Types

_types.TypeDecorator.process_result_value()

Return type:

Optional[Any]

Parameters:
process_result_value(value, dialect)[source]

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 (Optional[bytes]) – Data to operate upon, of any type expected by this method in the subclass. Can be None.

  • dialect (Dialect) – the Dialect in use.

  • value (bytes | None)

  • dialect (Dialect)

Return type:

Any | None

See also

Augmenting Existing Types

_types.TypeDecorator.process_bind_param()

Return type:

Optional[Any]

Parameters:
advanced_alchemy.types.Bool

alias of _OracleAwareBoolean

class advanced_alchemy.types.DateTimeUTC[source]

Bases: TypeDecorator

Timezone Aware DateTime.

Ensure UTC is stored in the database and that TZ aware dates are returned for all dialects.

impl = DateTime(timezone=True)
cache_ok = 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))))

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

Added 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 Compilation Caching

property python_type: type[datetime]

Return the Python type object expected to be returned by instances of this type, if known.

Basically, for those types which enforce a return type, or are known across the board to do such for all common DBAPIs (like int for example), will return that type.

If a return type is not defined, raises NotImplementedError.

Note that any type also accommodates NULL in SQL which means you can also get back None from any type in practice.

process_bind_param(value, dialect)[source]

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:
Return type:

datetime | None

See also

Augmenting Existing Types

_types.TypeDecorator.process_result_value()

Return type:

Optional[datetime]

Parameters:
process_result_value(value, dialect)[source]

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:
Return type:

datetime | None

See also

Augmenting Existing Types

_types.TypeDecorator.process_bind_param()

Return type:

Optional[datetime]

Parameters:
class advanced_alchemy.types.EncryptedString[source]

Bases: TypeDecorator

SQLAlchemy TypeDecorator for storing encrypted string values in a database.

This type provides transparent encryption/decryption of string values using the specified backend. It extends sqlalchemy.types.TypeDecorator and implements String as its underlying type.

Parameters:

  • key (str | bytes | Callable[[], str | bytes] | None) – The encryption key. Can be a string, bytes, or callable returning either. Defaults to os.urandom(32).

  • backend (Type[EncryptionBackend] | None) – The encryption backend class to use. Defaults to FernetBackend.

  • length (int | None) – The length of the unencrypted string. This is used for documentation and validation purposes only, as encrypted strings will be longer.

  • **kwargs (Any | None) – Additional arguments passed to the underlying String type.

key

The encryption key.

Type:

str | bytes | Callable[[], str | bytes]

backend

The encryption backend instance.

Type:

EncryptionBackend

length

The unencrypted string length.

Type:

int | None

impl

alias of String

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

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

Added 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 Compilation Caching

__init__(key=DEFAULT_ENCRYPTION_KEY, backend=FernetBackend, length=None, **kwargs)[source]

Initializes the EncryptedString TypeDecorator.

Parameters:

  • key (str | bytes | Callable[[], str | bytes] | None) – The encryption key. Can be a string, bytes, or callable returning either. Defaults to os.urandom(32).

  • backend (Type[EncryptionBackend] | None) – The encryption backend class to use. Defaults to FernetBackend.

  • length (int | None) – The length of the unencrypted string. This is used for documentation and validation purposes only.

  • **kwargs (Any | None) – Additional arguments passed to the underlying String type.

__repr__()[source]

Return a reconstructable representation of the type.

Uses type(self).__name__ so subclasses (e.g. EncryptedText) render their own name; this keeps Alembic autogenerate from reconstructing the wrong column type.

Return type:

str

property python_type: type[str]

Returns the Python type for this type decorator.

Returns:

The Python string type.

Return type:

Type[str]

load_dialect_impl(dialect)[source]

Loads the appropriate dialect implementation based on the database dialect.

Note: The actual column length will be larger than the specified length due to encryption overhead. For most encryption methods, the encrypted string will be approximately 1.35x longer than the original.

Parameters:
Returns:

The dialect-specific type descriptor.

Return type:

Any

process_bind_param(value, dialect)[source]

Processes the value before binding it to the SQL statement.

This method encrypts the value using the specified backend and validates length if specified.

Parameters:

  • value (Any) – The value to process.

  • dialect (Dialect) – The SQLAlchemy dialect.

  • value (Any)

  • dialect (Dialect)

Raises:

IntegrityError – If the unencrypted value exceeds the maximum length.

Returns:

The encrypted value or None if the input is None.

Return type:

str | None

process_result_value(value, dialect)[source]

Processes the value after retrieving it from the database.

This method decrypts the value using the specified backend.

Parameters:

  • value (Any) – The value to process.

  • dialect (Dialect) – The SQLAlchemy dialect.

  • value (Any)

  • dialect (Dialect)

Returns:

The decrypted value or None if the input is None.

Return type:

str | None

mount_vault()[source]

Mounts the vault with the encryption key.

For a static key the cipher is derived once and reused. For a callable key it is re-resolved on every call so rotated keys take effect.

Return type:

None

bind_expression(bindparam)[source]

Wraps the bound parameter with the backend’s server-side encryption, if any.

For client-side backends the parameter is returned unchanged so it compiles to a plain placeholder; server-side backends wrap it in a database encryption function.

Parameters:

  • bindparam (Any) – The bound parameter element.

  • bindparam (Any)

Returns:

The wrapped SQL expression for server-side backends, otherwise the bound parameter.

Return type:

Any

column_expression(column)[source]

Wraps the column read with the backend’s server-side decryption, if any.

For client-side backends the column is returned unchanged; server-side backends wrap it in a database decryption function.

Parameters:

  • column (Any) – The column element being read.

  • column (Any)

Returns:

The wrapped SQL expression for server-side backends, otherwise the column.

Return type:

Any

class advanced_alchemy.types.EncryptedText[source]

Bases: EncryptedString

SQLAlchemy TypeDecorator for storing encrypted text/CLOB values in a database.

This type provides transparent encryption/decryption of text values using the specified backend. It extends EncryptedString and implements Text as its underlying type. This is suitable for storing larger encrypted text content compared to EncryptedString.

Parameters:

  • key (str | bytes | Callable[[], str | bytes] | None) – The encryption key. Can be a string, bytes, or callable returning either. Defaults to os.urandom(32).

  • backend (Type[EncryptionBackend] | None) – The encryption backend class to use. Defaults to FernetBackend.

  • **kwargs (Any | None) – Additional arguments passed to the underlying String type.

impl

alias of Text

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

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

Added 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 Compilation Caching

load_dialect_impl(dialect)[source]

Loads the appropriate dialect implementation for Text type.

Parameters:
Returns:

The dialect-specific Text type descriptor.

Return type:

Any

class advanced_alchemy.types.EncryptionBackend[source]

Bases: ABC

Abstract base class for encryption backends.

This class defines the interface that all encryption backends must implement. Concrete implementations should provide the actual encryption/decryption logic.

passphrase

The encryption passphrase used by the backend.

Type:

bytes

mount_vault(key)[source]

Mounts the vault with the provided encryption key.

Parameters:

  • key (str | bytes) – The encryption key used to initialize the backend.

  • key (Union[str, bytes])

Return type:

None

abstractmethod init_engine(key)[source]

Initializes the encryption engine with the provided key.

Parameters:
Raises:

NotImplementedError – If the method is not implemented by the subclass.

Return type:

None

abstractmethod encrypt(value)[source]

Encrypts the given value.

Parameters:

  • value (Any) – The value to encrypt.

  • value (Any)

Returns:

The encrypted value.

Return type:

str

Raises:

NotImplementedError – If the method is not implemented by the subclass.

abstractmethod decrypt(value)[source]

Decrypts the given value.

Parameters:

  • value (Any) – The value to decrypt.

  • value (Any)

Returns:

The decrypted value.

Return type:

str

Raises:

NotImplementedError – If the method is not implemented by the subclass.

bind_expression(bindvalue)[source]

Returns a SQL expression that encrypts the bound parameter server-side.

Client-side backends (which encrypt in encrypt()) return None so the bound value is stored verbatim. Server-side backends override this to wrap the bind parameter in a database encryption function.

Parameters:

  • bindvalue (Any) – The bound parameter element.

  • bindvalue (Any)

Returns:

The wrapped SQL expression, or None for client-side backends.

Return type:

Any | None

column_expression(column)[source]

Returns a SQL expression that decrypts the column server-side.

Client-side backends (which decrypt in decrypt()) return None so the stored value is read verbatim. Server-side backends override this to wrap the column in a database decryption function.

Parameters:

  • column (Any) – The column element being read.

  • column (Any)

Returns:

The wrapped SQL expression, or None for client-side backends.

Return type:

Any | None

class advanced_alchemy.types.FernetBackend[source]

Bases: EncryptionBackend

Fernet-based encryption backend.

This backend uses the Python cryptography library’s Fernet implementation for encryption/decryption operations. Provides authenticated symmetric encryption (AES-128-CBC + HMAC-SHA256). The key is derived from the provided passphrase with a single SHA256 digest.

key

The base64-encoded key used for encryption and decryption.

Type:

bytes

fernet

The Fernet instance used for encryption/decryption.

Type:

cryptography.fernet.Fernet

__init__()[source]

Initializes the Fernet backend.

Raises:

MissingDependencyError – If the cryptography package is not installed.

Return type:

None

mount_vault(key)[source]

Mounts the vault with the provided encryption key.

This method hashes the key using SHA256 before initializing the engine.

Parameters:
Return type:

None

init_engine(key)[source]

Initializes the Fernet engine with the provided key.

Parameters:
Return type:

None

encrypt(value)[source]

Encrypts the given value using Fernet.

Parameters:

  • value (Any) – The value to encrypt.

  • value (Any)

Returns:

The encrypted value.

Return type:

str

decrypt(value)[source]

Decrypts the given value using Fernet.

Parameters:

  • value (Any) – The value to decrypt.

  • value (Any)

Returns:

The decrypted value.

Return type:

str

class advanced_alchemy.types.FileObject[source]

Bases: object

Represents file metadata during processing using a dataclass structure.

This class provides a unified interface for handling file metadata and operations across different storage backends.

Content or a source path can optionally be provided during initialization via kwargs, store it internally, and add save/save_async methods to persist this pending data using the configured backend.

__init__(backend, filename, to_filename=None, content_type=None, size=None, last_modified=None, checksum=None, etag=None, version_id=None, metadata=None, source_path=None, content=None)[source]

Perform post-initialization validation and setup.

Handles default path, content type guessing, backend protocol inference, and processing of ‘content’ or ‘source_path’ from extra kwargs.

Raises:

ValueError – If filename is not provided, size is negative, backend/protocol mismatch, or both ‘content’ and ‘source_path’ are provided.

__repr__()[source]

Return a string representation of the FileObject.

Return type:

str

__eq__(other)[source]

Check equality based on filename and backend key.

Parameters:
Returns:

True if the objects are equal, False otherwise.

Return type:

bool

__hash__()[source]

Return a hash based on filename and backend key.

Return type:

int

property backend: StorageBackend
property filename: str
property content_type: str
property protocol: str
property path: str
property has_pending_data: bool
property metadata: dict[str, Any]
property size: int | None
property last_modified: float | None
property checksum: str | None
property etag: str | None
property version_id: str | None
update_metadata(metadata)[source]

Update the file metadata.

Parameters:

  • metadata (dict[str, typing.Any]) – New metadata to merge with existing metadata.

  • metadata (dict[str, TypeAliasForwardRef('typing.Any')])

Return type:

None

to_dict()[source]

Convert FileObject to a dictionary for storage or serialization.

Note: The ‘backend’ attribute is intentionally excluded as it’s often

not serializable or relevant for storage representations. The ‘extra’ dict is included.

Returns:

A dictionary representation of the file information.

Return type:

dict[str, typing.Any]

get_content(*, options=None)[source]

Get the file content from the storage backend.

Parameters:

  • options – Optional backend-specific options.

  • options (Optional[dict[str, Any]])

Returns:

The file content.

Return type:

bytes

async get_content_async(*, options=None)[source]

Get the file content from the storage backend asynchronously.

Parameters:

  • options – Optional backend-specific options.

  • options (Optional[dict[str, Any]])

Returns:

The file content.

Return type:

bytes

sign(*, expires_in=None, for_upload=False)[source]

Generate a signed URL for the file.

Parameters:

  • expires_in – Optional expiration time in seconds.

  • for_upload – Whether the URL is for upload.

  • expires_in (Optional[int])

  • for_upload (bool)

Raises:

RuntimeError – If no signed URL is generated.

Returns:

The signed URL.

Return type:

str

async sign_async(*, expires_in=None, for_upload=False)[source]

Generate a signed URL for the file asynchronously.

Parameters:

  • expires_in – Optional expiration time in seconds.

  • for_upload – Whether the URL is for upload.

  • expires_in (Optional[int])

  • for_upload (bool)

Returns:

The signed URL.

Return type:

str

Raises:

RuntimeError – If no signed URL is generated.

delete()[source]

Delete the file from storage.

Raises:

RuntimeError – If no backend is configured or path is missing.

Return type:

None

async delete_async()[source]

Delete the file from storage asynchronously.

Return type:

None

save(data=None, *, use_multipart=None, chunk_size=5 * 1024 * 1024, max_concurrency=12)[source]

Save data to the storage backend using this FileObject’s metadata.

If data is provided, it is used directly. If data is None, checks internal source_content or source_path. Clears pending attributes after successful save.

Parameters:
Return type:

FileObject

Returns:

The updated FileObject instance returned by the backend.

Raises:

TypeError – If trying to save async data synchronously.

async save_async(data=None, *, use_multipart=None, chunk_size=5 * 1024 * 1024, max_concurrency=12)[source]

Save data to the storage backend asynchronously.

If data is provided, it is used directly. If data is None, checks internal source_content or source_path. Clears pending attributes after successful save. Uses asyncio.to_thread for reading source_path if backend doesn’t handle Path directly.

Parameters:
Return type:

FileObject

Returns:

The updated FileObject instance returned by the backend.

Raises:

TypeError – If trying to save sync data asynchronously.

classmethod __get_pydantic_core_schema__(source_type, handler)[source]

Get the Pydantic core schema for FileObject.

This method defines how Pydantic should validate and serialize FileObject instances. It creates a schema that validates dictionaries with the required fields and converts them to FileObject instances.

Raises:

MissingDependencyError – If Pydantic is not installed when this method is called.

Parameters:
Return type:

Union[InvalidSchema, AnySchema, NoneSchema, BoolSchema, IntSchema, FloatSchema, DecimalSchema, StringSchema, BytesSchema, DateSchema, TimeSchema, DatetimeSchema, TimedeltaSchema, LiteralSchema, MissingSentinelSchema, EnumSchema, IsInstanceSchema, IsSubclassSchema, CallableSchema, ListSchema, TupleSchema, SetSchema, FrozenSetSchema, GeneratorSchema, DictSchema, AfterValidatorFunctionSchema, BeforeValidatorFunctionSchema, WrapValidatorFunctionSchema, PlainValidatorFunctionSchema, WithDefaultSchema, NullableSchema, UnionSchema, TaggedUnionSchema, ChainSchema, LaxOrStrictSchema, JsonOrPythonSchema, TypedDictSchema, ModelFieldsSchema, ModelSchema, DataclassArgsSchema, DataclassSchema, ArgumentsSchema, ArgumentsV3Schema, CallSchema, CustomErrorSchema, JsonSchema, UrlSchema, MultiHostUrlSchema, DefinitionsSchema, DefinitionReferenceSchema, UuidSchema, ComplexSchema]

Returns:

A Pydantic core schema for FileObject

class advanced_alchemy.types.HashedOneTimeCode[source]

Bases: object

Read-side wrapper over a stored one-time code’s hash and lifecycle state.

verify only succeeds while the code is still redeemable (not expired, not used, and not locked out). State transitions return a new instance to assign back to the column; commit it to persist single-use and attempt tracking.

__init__(hash_string, backend, *, expires_at=None, used_at=None, attempts=0, max_attempts=None)[source]

Initialize the wrapper.

Parameters:

  • hash_string – The stored hash of the code.

  • backend – The hashing backend used to verify candidates.

  • expires_at – POSIX timestamp after which the code is expired, or None for no expiry.

  • used_at – POSIX timestamp when the code was redeemed, or None if unused.

  • attempts – Number of failed verification attempts recorded so far.

  • max_attempts – Attempt ceiling after which the code locks, or None for no limit.

hash_string
expires_at
used_at
attempts
property is_expired: bool

True if the code has an expiry and it has passed.

property is_used: bool

True if the code has already been redeemed.

property is_locked: bool

True if failed attempts have reached max_attempts.

property is_redeemable: bool

True if the code can still be redeemed (not expired, used, or locked).

verify(code)[source]

Return True only if code matches and the code is still redeemable.

This does not mutate state. Use redeem() to also produce the updated value to persist, or pair with consume()/register_failure().

Parameters:

  • code – The candidate code.

  • code (Union[str, bytes])

Returns:

True on a match of a still-redeemable code, otherwise False.

Return type:

bool

redeem(code)[source]

Verify code and return the updated state to persist.

On success the returned value is marked used (single-use); on a redeemable failure the attempt counter is incremented. Assign the returned wrapper back to the column and commit to enforce single-use, expiry, and attempt limits on the next load.

Parameters:

  • code – The candidate code.

  • code (Union[str, bytes])

Returns:

A (ok, new_state) tuple. new_state is unchanged when the code was already expired, used, or locked.

Return type:

tuple[bool, HashedOneTimeCode]

consume()[source]

Return a copy marked used now. Assign back and commit to enforce single-use.

Return type:

HashedOneTimeCode

register_failure()[source]

Return a copy with the failed-attempt counter incremented. Assign back and commit.

Return type:

HashedOneTimeCode

class advanced_alchemy.types.HashedPassword[source]

Bases: object

Wrapper class for a hashed password.

This class holds the hash string and provides a method to verify a plain text password against it.

__init__(hash_string, backend)[source]

Initialize a HashedPassword object.

Parameters:

  • hash_string (str) – The hash string.

  • backend (HashingBackend) – The hashing backend to use for verification.

  • hash_string (str)

  • backend (HashingBackend)

Return type:

None

verify(plain_password)[source]

Verify a plain text password against this hash.

Parameters:

  • plain_password – The plain text password to verify.

  • plain_password (Union[str, bytes])

Returns:

True if the password matches the hash, False otherwise.

Return type:

bool

verify_and_update(plain_password)[source]

Verify a password and return a fresh hash when the stored one is stale.

Parameters:

  • plain_password – The plain text password to verify.

  • plain_password (Union[str, bytes])

Returns:

(False, None) if the password does not match; (True, None) if it matches and no rehash is needed; (True, new_hash) if it matches but the stored hash should be replaced with the returned value.

Return type:

tuple[bool, Optional[str]]

class advanced_alchemy.types.MutableList[source]

Bases: MutableList[T]

A list type that implements Mutable.

The MutableList object implements a list that will emit change events to the underlying mapping when the contents of the list are altered, including when values are added or removed.

This is a replication of default Mutablelist provide by SQLAlchemy. The difference here is the properties _removed which keep every element removed from the list in order to be able to delete them after commit and keep them when session rolled back.

__init__(*args, **kwargs)[source]
Parameters:
Return type:

None

classmethod coerce(key, value)[source]

Convert plain list to instance of this class.

Return type:

typing.Any

Parameters:
__setitem__(index, value)[source]

Detect list set events and emit change events.

Return type:

None

Parameters:
__delitem__(index)[source]

Detect list del events and emit change events.

Return type:

None

Parameters:

index (Any)

pop(*arg)[source]

Remove and return item at index (default last).

Raises IndexError if list is empty or index is out of range.

Return type:

TypeVar(T, bound= Any)

Parameters:

arg (Any)

append(x)[source]

Append object to the end of the list.

Return type:

None

Parameters:

x (Any)

extend(x)[source]

Extend list by appending elements from the iterable.

Return type:

None

Parameters:

x (Any)

insert(i, x)[source]

Insert object before index.

Return type:

None

Parameters:
remove(i)[source]

Remove first occurrence of value.

Raises ValueError if the value is not present.

Return type:

None

Parameters:

i (T)

clear()[source]

Remove all items from list.

Return type:

None

sort(**kw)[source]

Sort the list in ascending order and return None.

The sort is in-place (i.e. the list itself is modified) and stable (i.e. the order of two equal elements is maintained).

If a key function is given, apply it once to each list item and sort them, ascending or descending, according to their function values.

The reverse flag can be set to sort in descending order.

Return type:

None

Parameters:

kw (Any)

reverse()[source]

Reverse IN PLACE.

Return type:

None

class advanced_alchemy.types.OneTimeCode[source]

Bases: TypeDecorator

Stores a transient one-time code hashed, with expiry and single-use state in JSON.

The column holds {"hash", "expires_at", "used_at", "attempts"} so the full lifecycle lives in one value: an expired or already-redeemed code fails verification, and the code can lock after max_attempts failures. Reads return a HashedOneTimeCode. Requires an explicit backend.

impl = JSON()
cache_ok = 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))))

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

Added 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 Compilation Caching

__init__(backend, ttl_seconds=None, max_attempts=3, **kwargs)[source]

Initialize the OneTimeCode type.

Single-use is always enforced (a successful redemption marks the code used); max_attempts governs how many wrong guesses lock the code before then.

Parameters:

  • backend – The hashing backend used to hash and verify codes.

  • ttl_seconds – Lifetime of a written code in seconds, or None for no expiry.

  • max_attempts – Wrong-guess ceiling after which a code locks. Defaults to 3; pass None to disable the lockout (the code stays single-use and still expires).

  • **kwargs – Additional arguments passed to the underlying JSON type.

__repr__()[source]

Return a reconstructable representation for Alembic autogenerate.

Return type:

str

property python_type: type[dict[str, TypeAliasForwardRef('typing.Any')]]

Returns the Python type for this type decorator.

process_bind_param(value, dialect)[source]

Hash a new code or serialize an updated wrapper to the stored JSON shape.

Parameters:
Returns:

The JSON-serializable state dict, or None.

Return type:

Optional[dict[str, Any]]

process_result_value(value, dialect)[source]

Wrap the stored JSON state in a HashedOneTimeCode.

Parameters:

  • value – The stored JSON object, or None.

  • dialect – The SQLAlchemy dialect.

  • value (Any)

  • dialect (Any)

Returns:

A HashedOneTimeCode over the stored state, or None.

Return type:

Optional[HashedOneTimeCode]

class advanced_alchemy.types.PasswordHash[source]

Bases: TypeDecorator

SQLAlchemy TypeDecorator for storing hashed passwords in a database.

This type provides transparent hashing of password values using the specified backend. It extends sqlalchemy.types.TypeDecorator and implements String as its underlying type.

impl

alias of String

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

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

Added 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 Compilation Caching

__init__(backend, length=255)[source]

Initialize the PasswordHash TypeDecorator.

Parameters:

  • backend (HashingBackend) – The hashing backend class to use

  • length (int) – The maximum length of the hash string. Defaults to 255.

  • backend (HashingBackend)

  • length (int)

Return type:

None

__repr__()[source]

Return a reconstructable representation of the type.

Uses type(self).__name__ so subclasses (e.g. OneTimeCode) render their own name; this keeps Alembic autogenerate from reconstructing the wrong type.

Return type:

str

property python_type: type[str]

Returns the Python type for this type decorator.

Returns:

The Python string type.

process_bind_param(value, dialect)[source]

Process the value before binding it to the SQL statement.

This method hashes the value using the specified backend. If the backend returns a SQLAlchemy FunctionElement (for DB-side hashing), it is returned directly. Otherwise, the hashed string is returned.

Parameters:

  • value – The value to process.

  • dialect – The SQLAlchemy dialect.

  • value (Any)

  • dialect (Any)

Returns:

The hashed string, a SQLAlchemy FunctionElement, or None.

Return type:

Union[str, FunctionElement[str], None]

process_result_value(value, dialect)[source]

Process the value after retrieving it from the database.

This method wraps the hash string in a HashedPassword object.

Parameters:

  • value – The value to process.

  • dialect – The SQLAlchemy dialect.

  • value (Any)

  • dialect (Any)

Returns:

A HashedPassword object or None if the input is None.

Return type:

Union[HashedPassword, None]

compare_value(column, plain_password)[source]

Generate a SQLAlchemy expression for comparing a column with a plain text password.

This is an optional advanced extension point for backends that can compare a hash server-side. All shipped backends (Argon2, Passlib, Pwdlib) raise NotImplementedError; the standard verify flow is client-side via HashedPassword.verify() and does not use this.

Parameters:

  • column – The SQLAlchemy column to compare.

  • plain_password – The plain text password to compare against.

  • column (ColumnElement[str])

  • plain_password (Union[str, bytes])

Returns:

A SQLAlchemy binary expression for the comparison.

Raises:

NotImplementedError – For all shipped backends.

Return type:

BinaryExpression[bool]

class advanced_alchemy.types.StorageBackend[source]

Bases: ABC

Unified protocol for storage backend implementations supporting both sync and async operations.

driver: str

The name of the storage backend.

protocol: str

The protocol used by the storage backend.

__init__(key, fs, **kwargs)[source]

Initialize the storage backend.

Parameters:

  • key (str) – The key of the backend instance

  • fs (Any) – The filesystem or storage client

  • **kwargs (Any) – Additional keyword arguments

  • key (str)

  • fs (Any)

  • kwargs (Any)

Return type:

None

key: str

The key of the backend instance.

abstractmethod get_content(path, *, options=None)[source]

Get the content of a file.

Parameters:
Returns:

The file content

Return type:

bytes

abstractmethod async get_content_async(path, *, options=None)[source]

Get the content of a file asynchronously.

Parameters:
Returns:

The file content

Return type:

bytes

abstractmethod save_object(file_object, data, *, use_multipart=None, chunk_size=5 * 1024 * 1024, max_concurrency=12)[source]

Store a file using information from a FileObject.

Parameters:

  • file_object – A FileObject instance containing metadata like path, content_type.

  • data – The file data to store.

  • use_multipart – Whether to use multipart upload.

  • chunk_size – Size of chunks for multipart upload.

  • max_concurrency – Maximum number of concurrent uploads.

  • file_object (FileObject)

  • data (DataLike)

  • use_multipart (Optional[bool])

  • chunk_size (int)

  • max_concurrency (int)

Returns:

The stored file object, potentially updated with backend info (size, etag, etc.).

Return type:

FileObject

abstractmethod async save_object_async(file_object, data, *, use_multipart=None, chunk_size=5 * 1024 * 1024, max_concurrency=12)[source]

Store a file asynchronously using information from a FileObject.

Parameters:
Returns:

The stored file object, potentially updated with backend info (size, etag, etc.).

Return type:

FileObject

abstractmethod delete_object(paths)[source]

Delete one or more files.

Parameters:
Return type:

None

abstractmethod async delete_object_async(paths)[source]

Delete one or more files asynchronously.

Parameters:
Return type:

None

abstractmethod sign(paths, *, expires_in=None, for_upload=False)[source]

Generate a signed URL for one or more files.

Parameters:
Returns:

The signed URL

Return type:

Union[str, list[str]]

abstractmethod async sign_async(paths, *, expires_in=None, for_upload=False)[source]

Generate a signed URL for one or more files asynchronously.

Parameters:
Returns:

The signed URL

Return type:

Union[str, list[str]]

class advanced_alchemy.types.StorageRegistry[source]

Bases: object

A provider for creating and managing threaded portals.

__init__(json_serializer=encode_json, json_deserializer=decode_json, default_backend=DEFAULT_BACKEND)[source]

Initialize the PortalProvider.

set_default_backend(default_backend)[source]

Set the default storage backend.

Parameters:
Return type:

None

is_registered(key)[source]

Check if a storage backend is registered in the registry.

Parameters:

  • key (str) – The key of the storage backend

  • key (str)

Returns:

True if the storage backend is registered, False otherwise.

Return type:

bool

get_backend(key)[source]

Retrieve a configured storage backend from the registry.

Returns:

The storage backend associaStorageBackendiven key.

Return type:

StorageBackend

Raises:

ImproperConfigurationError – If no storage backend is registered with the given key.

Parameters:

key (str)

register_backend(value, key=None)[source]
Overloads:

  • self, value (str) → None

  • self, value (str), key (None) → None

  • self, value (str), key (str) → None

  • self, value (StorageBackend), key (None) → None

  • self, value (StorageBackend), key (str) → None

Parameters:
Return type:

None

Register a new storage backend in the registry.

Parameters:

  • value – The storage backend to register.

  • key – The key to register the storage backend with.

  • value (Union[StorageBackend, str])

  • key (Optional[str])

Raises:

ImproperConfigurationError – If a string value is provided without a key.

Return type:

None

unregister_backend(key)[source]

Unregister a storage backend from the registry.

Return type:

None

Parameters:

key (str)

clear_backends()[source]

Clear the registry.

Return type:

None

registered_backends()[source]

Return a list of all registered keys.

Return type:

list[str]

class advanced_alchemy.types.StoredObject[source]

Bases: TypeDecorator

Custom SQLAlchemy type for storing single or multiple file metadata.

Stores file metadata in JSONB and handles file validation, processing, and storage operations through a configured storage backend.

impl = JSON()
cache_ok = 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))))

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

Added 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 Compilation Caching

property python_type: type[FileObject | list[FileObject] | set[FileObject] | MutableList[FileObject] | None]

Specifies the Python type used, accounting for the multiple flag.

property backend: StorageBackend

Resolves and returns the storage backend instance.

property storage_key: str

Returns the storage key from the resolved backend.

__init__(backend, multiple=False, *args, **kwargs)[source]

Initialize StoredObject type.

Parameters:

  • backend (Union[str, StorageBackend]) – Key to retrieve the backend or from the storage registry or storage backend to use.

  • multiple (bool) – If True, stores a list of files; otherwise, a single file.

  • *args (typing.Any) – Additional positional arguments for TypeDecorator.

  • **kwargs (typing.Any) – Additional keyword arguments for TypeDecorator.

  • backend (str | StorageBackend)

  • multiple (bool)

  • args (Any)

  • kwargs (Any)

Return type:

None

multiple: bool
__repr__()[source]

Return a string representation of the StoredObject.

Return type:

str

process_bind_param(value, dialect)[source]

Convert FileObject(s) to JSON representation for the database.

Injects the configured backend into the FileObject before conversion.

Note: This method expects an already processed FileInfo or its dict representation.

Use handle_upload() or handle_upload_async() for processing raw uploads.

Parameters:

  • value – The value to process

  • dialect – The SQLAlchemy dialect

  • value (Optional[FileObjectOrList])

  • dialect (Any)

Raises:

TypeError – If the input value is not a FileObject or a list of FileObjects.

Returns:

A dictionary representing the file metadata, or None if the input value is None.

Return type:

Optional[Union[dict[str, Any], list[dict[str, Any]]]]

process_result_value(value, dialect)[source]

Convert database JSON back to FileObject or MutableList[FileObject].

Parameters:
Raises:

TypeError – If the input value is not a list of dicts.

Returns:

FileObject or MutableList[FileObject] or None.

Return type:

Optional[FileObjectOrList]

class advanced_alchemy.types.TOTPProvider[source]

Bases: object

Read-side wrapper exposing pyotp operations over a decrypted secret.

The secret must be a valid base32 string. pyotp decodes it lazily, so a malformed secret raises binascii.Error on the first call that decodes it (now/verify).

__init__(secret, *, digits=6, interval=30, digest=None, issuer=None)[source]

Initialize the provider.

Parameters:

  • secret – The decrypted base32 secret.

  • digits – Number of digits in generated codes. Defaults to 6.

  • interval – Time step in seconds. Defaults to 30.

  • digest – Hash algorithm passed to pyotp (None selects SHA1).

  • issuer – Default issuer used for provisioning URIs.

Raises:

MissingDependencyError – If the pyotp package is not installed.

property secret: str

The decrypted base32 secret.

now()[source]

Return the current TOTP code.

Return type:

str

Returns:

The zero-padded code for the current time step.

verify(code, valid_window=1)[source]

Verify a code, tolerating valid_window ticks of clock drift each way.

Parameters:

  • code (str) – The candidate code.

  • valid_window (int) – Number of time steps of drift accepted on each side. Defaults to 1.

  • code (str)

  • valid_window (int)

Return type:

bool

Returns:

True if the code is valid within the window, False otherwise.

provisioning_uri(name=None, issuer_name=None)[source]

Return an otpauth:// provisioning URI for QR enrollment.

Parameters:

  • name – The account name (e.g. the user’s email).

  • issuer_name – The issuer label; falls back to the type’s configured issuer.

  • name (Optional[str])

  • issuer_name (Optional[str])

Returns:

The provisioning URI string.

Return type:

str

class advanced_alchemy.types.TOTPSecret[source]

Bases: EncryptedString

Stores a TOTP shared secret encrypted at rest, returning a TOTPProvider on read.

Storage is identical to EncryptedString; the read side wraps the decrypted secret in a TOTPProvider so callers can generate and verify codes. Requires pyotp.

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

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

Added 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 Compilation Caching

__init__(key, backend=FernetBackend, digits=6, interval=30, digest=None, issuer=None, length=None, **kwargs)[source]

Initialize the TOTPSecret type.

Parameters:

  • key – The encryption key (str/bytes or a callable returning one). Required: unlike EncryptedString, this new type has no deprecated random default.

  • backend – The encryption backend class. Defaults to FernetBackend.

  • digits – Number of digits in generated codes. Defaults to 6.

  • interval – Time step in seconds. Defaults to 30.

  • digest – Hash algorithm passed to pyotp (None selects SHA1).

  • issuer – Default issuer used for provisioning URIs.

  • length – The unencrypted string length (documentation only).

  • **kwargs – Additional arguments passed to the underlying String type.

Raises:

MissingDependencyError – If the pyotp package is not installed.

__repr__()[source]

Return a reconstructable representation including the TOTP parameters.

Overrides EncryptedString so Alembic autogenerate reconstructs the type with its configured digits/interval/digest/issuer rather than the defaults.

Return type:

str

process_bind_param(value, dialect)[source]

Accept a raw secret string or a TOTPProvider, then encrypt as usual.

Parameters:

  • value – The secret string or a TOTPProvider.

  • dialect – The SQLAlchemy dialect.

  • value (Any)

  • dialect (Dialect)

Returns:

The encrypted secret, or None.

Return type:

Union[str, None]

process_result_value(value, dialect)[source]

Decrypt the stored secret and wrap it in a TOTPProvider.

Parameters:

  • value – The stored encrypted value.

  • dialect – The SQLAlchemy dialect.

  • value (Any)

  • dialect (Dialect)

Returns:

A TOTPProvider over the decrypted secret, or None.

Return type:

Optional[TOTPProvider]

class advanced_alchemy.types.Vector[source]

Bases: TypeDecorator

Dialect-aware fixed-dimension vector column.

Parameters:
impl

alias of JSON

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

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

Added 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 Compilation Caching

property python_type: type[list[float]]

Return the Python type object expected to be returned by instances of this type, if known.

Basically, for those types which enforce a return type, or are known across the board to do such for all common DBAPIs (like int for example), will return that type.

If a return type is not defined, raises NotImplementedError.

Note that any type also accommodates NULL in SQL which means you can also get back None from any type in practice.

__init__(dim, *, storage_format='FLOAT32')[source]

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.

Parameters:

  • dim (int)

  • storage_format (str)

Return type:

None

load_dialect_impl(dialect)[source]

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.

Return type:

TypeEngine

Parameters:

dialect (Dialect)

process_result_value(value, dialect)[source]

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 (Any) – Data to operate upon, of any type expected by this method in the subclass. Can be None.

  • dialect (Dialect) – the Dialect in use.

  • value (Any)

  • dialect (Dialect)

Return type:

list[float] | None

See also

Augmenting Existing Types

_types.TypeDecorator.process_bind_param()

Return type:

Optional[list[float]]

Parameters:
class Comparator[source]

Bases: Comparator[list[float]]

Dialect-aware vector distance operators for similarity search.

Method names and semantics mirror pgvector’s SQLAlchemy comparator so existing pgvector-based queries can swap to Vector unchanged.

cosine_distance(other)[source]
Return type:

_VectorDistance

Parameters:

other (Sequence[float])

l2_distance(other)[source]
Return type:

_VectorDistance

Parameters:

other (Sequence[float])

l1_distance(other)[source]
Return type:

_VectorDistance

Parameters:

other (Sequence[float])

max_inner_product(other)[source]
Return type:

_VectorDistance

Parameters:

other (Sequence[float])

comparator_factory

alias of Comparator

advanced_alchemy.types.generate_one_time_code(length=6, *, digits_only=True)[source]

Generate a cryptographically secure random one-time code.

Parameters:

  • length (int) – Number of characters in the code. Defaults to 6.

  • digits_only (bool) – Use digits only when True (default); otherwise uppercase letters and digits.

  • length (int)

  • digits_only (bool)

Return type:

str

Returns:

A random one-time code string.

advanced_alchemy.types.generate_totp_secret(length=32)[source]

Generate a new random base32 TOTP secret.

Parameters:

  • length (int) – Number of base32 characters. Defaults to 32.

  • length (int)

Raises:

MissingDependencyError – If the pyotp package is not installed.

Return type:

str

Returns:

A base32-encoded secret suitable for an authenticator app.

class advanced_alchemy.types.encrypted_string.PGCryptoBackend[source]

Bases: EncryptionBackend

PostgreSQL pgcrypto-based encryption backend.

This backend uses PostgreSQL’s pgcrypto extension for encryption/decryption operations. Requires the pgcrypto extension to be installed in the database. Encryption and decryption run server-side via bind_expression() and column_expression(); the ASCII-armored ciphertext is stored as text so the column type is unchanged.

passphrase

The base64-encoded passphrase used for encryption and decryption.

Type:

str

init_engine(key)[source]

Initializes the pgcrypto engine with the provided key.

Parameters:
Return type:

None

encrypt(value)[source]

Returns the plaintext value to bind; encryption happens server-side.

Parameters:

  • value (Any) – The value to encrypt.

  • value (Any)

Returns:

The plaintext value passed to bind_expression().

Return type:

str

decrypt(value)[source]

Normalizes the server-decrypted value; decryption happens in SQL.

Parameters:
Returns:

The decrypted value.

Return type:

str

bind_expression(bindvalue)[source]

Wraps the bound parameter in armor(pgp_sym_encrypt(...)).

Parameters:

  • bindvalue (Any) – The bound parameter element.

  • bindvalue (Any)

Returns:

The SQL expression producing ASCII-armored ciphertext.

Return type:

Any

column_expression(column)[source]

Wraps the column in pgp_sym_decrypt(dearmor(...)).

Parameters:

  • column (Any) – The column element being read.

  • column (Any)

Returns:

The SQL expression producing the decrypted plaintext.

Return type:

Any

class advanced_alchemy.types.password_hash.argon2.Argon2Hasher[source]

Bases: HashingBackend

Hashing backend using Argon2 via the argon2-cffi library.

__init__(**kwargs)[source]

Initialize Argon2Backend.

Parameters:

  • **kwargs (Any) – Optional keyword arguments to pass to the argon2.PasswordHasher constructor. See argon2-cffi documentation for available parameters (e.g., time_cost, memory_cost, parallelism, hash_len, salt_len, type).

  • kwargs (Any)

Raises:

MissingDependencyError – If the argon2-cffi package is not installed.

Return type:

None

hash(value)[source]

Hash the password using Argon2.

Parameters:

  • value – The plain text password (will be encoded to UTF-8 if string).

  • value (Union[str, bytes])

Returns:

The Argon2 hash string.

Return type:

str

verify(plain, hashed)[source]

Verify a plain text password against an Argon2 hash.

Parameters:

  • plain (Any) – The plain text password (will be encoded to UTF-8 if string). Any other type returns False rather than raising.

  • hashed (str) – The Argon2 hash string to verify against.

  • plain (Any)

  • hashed (str)

Return type:

bool

Returns:

True if the password matches the hash, False otherwise.

needs_rehash(hashed)[source]

Return True if the hash was produced with weaker parameters than the current config.

Parameters:

  • hashed (str) – The stored Argon2 hash string.

  • hashed (str)

Return type:

bool

Returns:

True if the hash should be regenerated; False for an unparsable or foreign hash.

compare_expression(column, plain)[source]

Direct SQL comparison is not supported for Argon2.

Raises:

NotImplementedError – Always raised.

Parameters:

  • column (ColumnElement[str])

  • plain (Union[str, bytes])

Return type:

BinaryExpression[bool]

class advanced_alchemy.types.password_hash.passlib.PasslibHasher[source]

Bases: HashingBackend

Hashing backend using Passlib.

Relies on the passlib package being installed. Install with pip install passlib or uv pip install passlib.

__init__(context)[source]

Initialize PasslibBackend.

Parameters:

  • context (CryptContext) – The Passlib CryptContext to use for hashing and verification.

  • context (CryptContext)

Raises:

MissingDependencyError – If the passlib package is not installed.

Return type:

None

hash(value)[source]

Hash the given value using the Passlib context.

Parameters:

  • value – The plain text value to hash. Will be converted to string.

  • value (Union[str, bytes])

Returns:

The hashed string.

Return type:

str

verify(plain, hashed)[source]

Verify a plain text value against a hash using the Passlib context.

Parameters:

  • plain – The plain text value to verify. Will be converted to string.

  • hashed – The hash to verify against.

  • plain (Union[str, bytes])

  • hashed (str)

Returns:

True if the plain text matches the hash, False otherwise.

Return type:

bool

needs_rehash(hashed)[source]

Return True if the stored hash uses an outdated scheme or cost.

Parameters:

  • hashed (str) – The stored hash string.

  • hashed (str)

Return type:

bool

Returns:

True if the hash should be regenerated; False for an unparsable or foreign hash.

compare_expression(column, plain)[source]

Direct SQL comparison is not supported for Passlib.

Raises:

NotImplementedError – Always raised.

Return type:

BinaryExpression[bool]

Parameters:
class advanced_alchemy.types.password_hash.pwdlib.PwdlibHasher[source]

Bases: HashingBackend

Hashing backend using Pwdlib.

__init__(hasher)[source]

Initialize PwdlibBackend.

Parameters:

  • hasher (HasherProtocol) – The Pwdlib hasher to use for hashing and verification.

  • hasher (HasherProtocol)

Raises:

MissingDependencyError – If the pwdlib package is not installed.

Return type:

None

hash(value)[source]

Hash the given value using the Pwdlib context.

Parameters:

  • value – The plain text value to hash. Will be converted to string.

  • value (Union[str, bytes])

Returns:

The hashed string.

Return type:

str

verify(plain, hashed)[source]

Verify a plain text value against a hash using the Pwdlib context.

Parameters:

  • plain – The plain text value to verify. Will be converted to string.

  • hashed – The hash to verify against.

  • plain (Union[str, bytes])

  • hashed (str)

Returns:

True if the plain text matches the hash, False otherwise.

Return type:

bool

needs_rehash(hashed)[source]

Return True if the wrapped hasher reports the stored hash as stale.

Parameters:

  • hashed (str) – The stored hash string.

  • hashed (str)

Return type:

bool

Returns:

True if the hash should be regenerated; False for an unparsable or foreign hash.

compare_expression(column, plain)[source]

Direct SQL comparison is not supported for Pwdlib.

Raises:

NotImplementedError – Always raised.

Return type:

BinaryExpression[bool]

Parameters: