calibpipe.database.interfaces package#

Various interfaces for database access.

This module contains field, column, row and table interfaces to interact with a database. In general the module is tightly coupled to sqlalchemy though some parts have been preserved from an explicit dependency.

Module content:

types.py: Type definition for DB fields using sqlalchemy types and defining the specific case of the numpy.ndarray.
sql_metadata.py: Simple metadata variable definition for sqlalchemy.
sql_column_info.py: Container for column information, useful to e.g. define the corresponding LST fields at the DB level.
sql_table_info.py: Container for table information, used in particular to create sqlalchemy table objects from a list of column information.
hashable_row_data.py: Hashable container of a (single) row information (table_name + primary_key value). This can be (and is) used to index data using the row to which they belong and create a cache for data retrieved from the database.
table_handler.py: Container for functions to handle tables in the DB.
queries.py: Built-in queries that can be used to retrieve camera calibration data.

class calibpipe.database.interfaces.Boolean(create_constraint: bool = False, name: str | None = None, _create_events: bool = True, _adapted_from: SchemaType | None = None)[source]#

Bases: SchemaType, Emulated, TypeEngine

A bool datatype.

Boolean typically uses BOOLEAN or SMALLINT on the DDL side, and on the Python side deals in True or False.

The Boolean datatype currently has two levels of assertion that the values persisted are simple true/false values. For all backends, only the Python values None, True, False, 1 or 0 are accepted as parameter values. For those backends that don’t support a “native boolean” datatype, an option exists to also create a CHECK constraint on the target column

Changed in version 1.2: the Boolean datatype now asserts that incoming Python values are already in pure boolean form.

native = True#

__init__(create_constraint: bool = False, name: str | None = None, _create_events: bool = True, _adapted_from: SchemaType | None = None)[source]#

Construct a Boolean.

Parameters:

create_constraint¶ –
defaults to False. If the boolean is generated as an int/smallint, also create a CHECK constraint on the table that ensures 1 or 0 as a value.

Note

it is strongly recommended that the CHECK constraint have an explicit name in order to support schema-management concerns. This can be established either by setting the Boolean.name parameter or by setting up an appropriate naming convention; see Configuring Constraint Naming Conventions for background.

Changed in version 1.4: - this flag now defaults to False, meaning no CHECK constraint is generated for a non-native enumerated type.
name¶ – if a CHECK constraint is generated, specify the name of the constraint.

copy(**kw)[source]#

property python_type#

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.

literal_processor(dialect)[source]#

Return a conversion function for processing literal values that are to be rendered directly without using binds.

This function is used when the compiler makes use of the “literal_binds” flag, typically used in DDL generation as well as in certain scenarios where backends don’t accept bound parameters.

Returns a callable which will receive a literal Python value as the sole positional argument and will return a string representation to be rendered in a SQL statement.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.literal_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.literal_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_literal_param().

See also

Augmenting Existing Types

bind_processor(dialect)[source]#

Return a conversion function for processing bind values.

Returns a callable which will receive a bind parameter value as the sole positional argument and will return a value to send to the DB-API.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.bind_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.bind_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_bind_param().

See also

Augmenting Existing Types

Parameters:: dialect¶ – Dialect instance in use.

result_processor(dialect, coltype)[source]#

Return a conversion function for processing result row values.

Returns a callable which will receive a result row column value as the sole positional argument and will return a value to return to the user.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.result_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.result_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_result_value().

See also

Augmenting Existing Types

Parameters:

dialect¶ – Dialect instance in use.
coltype¶ – DBAPI coltype argument received in cursor.description.

class calibpipe.database.interfaces.SmallInteger[source]#

Bases: Integer

A type for smaller int integers.

Typically generates a SMALLINT in DDL, and otherwise acts like a normal Integer on the Python side.

class calibpipe.database.interfaces.Integer[source]#

Bases: HasExpressionLookup, TypeEngine

A type for int integers.

get_dbapi_type(dbapi)[source]#

Return the corresponding type object from the underlying DB-API, if any.

This can be useful for calling setinputsizes(), for example.

property python_type#

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.

literal_processor(dialect)[source]#

Return a conversion function for processing literal values that are to be rendered directly without using binds.

This function is used when the compiler makes use of the “literal_binds” flag, typically used in DDL generation as well as in certain scenarios where backends don’t accept bound parameters.

Returns a callable which will receive a literal Python value as the sole positional argument and will return a string representation to be rendered in a SQL statement.

Tip

To provide alternate behavior for _types.TypeEngine.literal_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_literal_param().

See also

Augmenting Existing Types

class calibpipe.database.interfaces.BigInteger[source]#

Bases: Integer

A type for bigger int integers.

Typically generates a BIGINT in DDL, and otherwise acts like a normal Integer on the Python side.

class calibpipe.database.interfaces.Float(precision: int | None = ..., asdecimal: Literal[False] = ..., decimal_return_scale: int | None = ...)[source]#

class calibpipe.database.interfaces.Float(precision: int | None = ..., asdecimal: Literal[True] = ..., decimal_return_scale: int | None = ...)

Bases: Numeric

Type representing floating point types, such as FLOAT or REAL.

This type returns Python float objects by default, unless the Float.asdecimal flag is set to True, in which case they are coerced to decimal.Decimal objects.

When a Float.precision is not provided in a _types.Float type some backend may compile this type as an 8 bytes / 64 bit float datatype. To use a 4 bytes / 32 bit float datatype a precision <= 24 can usually be provided or the _types.REAL type can be used. This is known to be the case in the PostgreSQL and MSSQL dialects that render the type as FLOAT that’s in both an alias of DOUBLE PRECISION. Other third party dialects may have similar behavior.

scale = None#

__init__(precision: int | None = None, asdecimal: Literal[False] = False, decimal_return_scale: int | None = None)[source]#

__init__(precision: int | None = None, asdecimal: Literal[True] = False, decimal_return_scale: int | None = None)

Construct a Float.

Parameters:

precision¶ –
the numeric precision for use in DDL CREATE TABLE. Backends should attempt to ensure this precision indicates a number of digits for the generic _sqltypes.Float datatype.
Note

For the Oracle Database backend, the _sqltypes.Float.precision parameter is not accepted when rendering DDL, as Oracle Database does not support float precision specified as a number of decimal places. Instead, use the Oracle Database-specific _oracle.FLOAT datatype and specify the _oracle.FLOAT.binary_precision parameter. This is new in version 2.0 of SQLAlchemy.

To create a database agnostic _types.Float that separately specifies binary precision for Oracle Database, use _types.TypeEngine.with_variant() as follows:
```
from sqlalchemy import Column
from sqlalchemy import Float
from sqlalchemy.dialects import oracle

Column(
    "float_data",
    Float(5).with_variant(oracle.FLOAT(binary_precision=16), "oracle"),
)
```
asdecimal¶ – the same flag as that of Numeric, but defaults to False. Note that setting this flag to True results in floating point conversion.
decimal_return_scale¶ – Default scale to use when converting from floats to Python decimals. Floating point values will typically be much longer due to decimal inaccuracy, and most floating point database types don’t have a notion of “scale”, so by default the float type looks for the first ten decimal places when converting. Specifying this value will override that length. Note that the MySQL float types, which do include “scale”, will use “scale” as the default for decimal_return_scale, if not otherwise specified.

result_processor(dialect, coltype)[source]#

Return a conversion function for processing result row values.

Returns a callable which will receive a result row column value as the sole positional argument and will return a value to return to the user.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.result_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.result_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_result_value().

See also

Augmenting Existing Types

Parameters:

dialect¶ – Dialect instance in use.
coltype¶ – DBAPI coltype argument received in cursor.description.

calibpipe.database.interfaces.Double[source]#: alias of DOUBLE_PRECISION

class calibpipe.database.interfaces.Numeric(precision: int | None = ..., scale: int | None = ..., decimal_return_scale: int | None = ..., asdecimal: Literal[True] = ...)[source]#

class calibpipe.database.interfaces.Numeric(precision: int | None = ..., scale: int | None = ..., decimal_return_scale: int | None = ..., asdecimal: Literal[False] = ...)

Bases: HasExpressionLookup, TypeEngine

Base for non-integer numeric types, such as NUMERIC, FLOAT, DECIMAL, and other variants.

The Numeric datatype when used directly will render DDL corresponding to precision numerics if available, such as NUMERIC(precision, scale). The Float subclass will attempt to render a floating-point datatype such as FLOAT(precision).

Numeric returns Python decimal.Decimal objects by default, based on the default value of True for the Numeric.asdecimal parameter. If this parameter is set to False, returned values are coerced to Python float objects.

The Float subtype, being more specific to floating point, defaults the Float.asdecimal flag to False so that the default Python datatype is float.

Note

When using a Numeric datatype against a database type that returns Python floating point values to the driver, the accuracy of the decimal conversion indicated by Numeric.asdecimal may be limited. The behavior of specific numeric/floating point datatypes is a product of the SQL datatype in use, the Python DBAPI in use, as well as strategies that may be present within the SQLAlchemy dialect in use. Users requiring specific precision/ scale are encouraged to experiment with the available datatypes in order to determine the best results.

__init__(precision: int | None = None, scale: int | None = None, decimal_return_scale: int | None = None, asdecimal: Literal[True] = True)[source]#

__init__(precision: int | None = None, scale: int | None = None, decimal_return_scale: int | None = None, asdecimal: Literal[False] = True)

Construct a Numeric.

Parameters:

precision¶ – the numeric precision for use in DDL CREATE TABLE.
scale¶ – the numeric scale for use in DDL CREATE TABLE.
asdecimal¶ – default True. Return whether or not values should be sent as Python Decimal objects, or as floats. Different DBAPIs send one or the other based on datatypes - the Numeric type will ensure that return values are one or the other across DBAPIs consistently.
decimal_return_scale¶ – Default scale to use when converting from floats to Python decimals. Floating point values will typically be much longer due to decimal inaccuracy, and most floating point database types don’t have a notion of “scale”, so by default the float type looks for the first ten decimal places when converting. Specifying this value will override that length. Types which do include an explicit “.scale” value, such as the base Numeric as well as the MySQL float types, will use the value of “.scale” as the default for decimal_return_scale, if not otherwise specified.

When using the Numeric type, care should be taken to ensure that the asdecimal setting is appropriate for the DBAPI in use - when Numeric applies a conversion from Decimal->float or float-> Decimal, this conversion incurs an additional performance overhead for all result columns received.

DBAPIs that return Decimal natively (e.g. psycopg2) will have better accuracy and higher performance with a setting of True, as the native translation to Decimal reduces the amount of floating- point issues at play, and the Numeric type itself doesn’t need to apply any further conversions. However, another DBAPI which returns floats natively will incur an additional conversion overhead, and is still subject to floating point data loss - in which case asdecimal=False will at least remove the extra conversion overhead.

get_dbapi_type(dbapi)[source]#

Return the corresponding type object from the underlying DB-API, if any.

This can be useful for calling setinputsizes(), for example.

literal_processor(dialect)[source]#

Return a conversion function for processing literal values that are to be rendered directly without using binds.

This function is used when the compiler makes use of the “literal_binds” flag, typically used in DDL generation as well as in certain scenarios where backends don’t accept bound parameters.

Returns a callable which will receive a literal Python value as the sole positional argument and will return a string representation to be rendered in a SQL statement.

Tip

To provide alternate behavior for _types.TypeEngine.literal_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_literal_param().

See also

Augmenting Existing Types

property python_type#

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.

bind_processor(dialect)[source]#

Return a conversion function for processing bind values.

Returns a callable which will receive a bind parameter value as the sole positional argument and will return a value to send to the DB-API.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.bind_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.bind_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_bind_param().

See also

Augmenting Existing Types

Parameters:: dialect¶ – Dialect instance in use.

result_processor(dialect, coltype)[source]#

Return a conversion function for processing result row values.

Returns a callable which will receive a result row column value as the sole positional argument and will return a value to return to the user.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.result_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.result_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_result_value().

See also

Augmenting Existing Types

Parameters:

dialect¶ – Dialect instance in use.
coltype¶ – DBAPI coltype argument received in cursor.description.

class calibpipe.database.interfaces.String(length: int | None = None, collation: str | None = None)[source]#

Bases: Concatenable, TypeEngine

The base for all string and character types.

In SQL, corresponds to VARCHAR.

The length field is usually required when the String type is used within a CREATE TABLE statement, as VARCHAR requires a length on most databases.

__init__(length: int | None = None, collation: str | None = None)[source]#

Create a string-holding type.

Parameters:

length¶ – optional, a length for the column for use in DDL and CAST expressions. May be safely omitted if no CREATE TABLE will be issued. Certain databases may require a length for use in DDL, and will raise an exception when the CREATE TABLE DDL is issued if a VARCHAR with no length is included. Whether the value is interpreted as bytes or characters is database specific.
collation¶ –
Optional, a column-level collation for use in DDL and CAST expressions. Renders using the COLLATE keyword supported by SQLite, MySQL, and PostgreSQL. E.g.:
```
>>> from sqlalchemy import cast, select, String
>>> print(select(cast("some string", String(collation="utf8"))))
{printsql}SELECT CAST(:param_1 AS VARCHAR COLLATE utf8) AS anon_1
```
Note

In most cases, the Unicode or UnicodeText datatypes should be used for a _schema.Column that expects to store non-ascii data. These datatypes will ensure that the correct types are used on the database.

literal_processor(dialect)[source]#

Return a conversion function for processing literal values that are to be rendered directly without using binds.

This function is used when the compiler makes use of the “literal_binds” flag, typically used in DDL generation as well as in certain scenarios where backends don’t accept bound parameters.

Returns a callable which will receive a literal Python value as the sole positional argument and will return a string representation to be rendered in a SQL statement.

Tip

To provide alternate behavior for _types.TypeEngine.literal_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_literal_param().

See also

Augmenting Existing Types

bind_processor(dialect: Dialect) → _BindProcessorType[str] | None[source]#

Return a conversion function for processing bind values.

Returns a callable which will receive a bind parameter value as the sole positional argument and will return a value to send to the DB-API.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.bind_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.bind_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_bind_param().

See also

Augmenting Existing Types

Parameters:: dialect¶ – Dialect instance in use.

result_processor(dialect: Dialect, coltype: object) → _ResultProcessorType[str] | None[source]#

Return a conversion function for processing result row values.

Returns a callable which will receive a result row column value as the sole positional argument and will return a value to return to the user.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.result_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.result_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_result_value().

See also

Augmenting Existing Types

Parameters:

dialect¶ – Dialect instance in use.
coltype¶ – DBAPI coltype argument received in cursor.description.

property python_type#

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.

get_dbapi_type(dbapi)[source]#

Return the corresponding type object from the underlying DB-API, if any.

This can be useful for calling setinputsizes(), for example.

class calibpipe.database.interfaces.Date[source]#

Bases: _RenderISO8601NoT, HasExpressionLookup, TypeEngine

A type for datetime.date() objects.

get_dbapi_type(dbapi)[source]#

Return the corresponding type object from the underlying DB-API, if any.

This can be useful for calling setinputsizes(), for example.

property python_type#

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.

literal_processor(dialect)[source]#

Return a conversion function for processing literal values that are to be rendered directly without using binds.

This function is used when the compiler makes use of the “literal_binds” flag, typically used in DDL generation as well as in certain scenarios where backends don’t accept bound parameters.

Returns a callable which will receive a literal Python value as the sole positional argument and will return a string representation to be rendered in a SQL statement.

Tip

To provide alternate behavior for _types.TypeEngine.literal_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_literal_param().

See also

Augmenting Existing Types

class calibpipe.database.interfaces.Time(timezone: bool = False)[source]#

Bases: _RenderISO8601NoT, HasExpressionLookup, TypeEngine

A type for datetime.time() objects.

__init__(timezone: bool = False)[source]#

get_dbapi_type(dbapi)[source]#

Return the corresponding type object from the underlying DB-API, if any.

This can be useful for calling setinputsizes(), for example.

property python_type#

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.

literal_processor(dialect)[source]#

Return a conversion function for processing literal values that are to be rendered directly without using binds.

This function is used when the compiler makes use of the “literal_binds” flag, typically used in DDL generation as well as in certain scenarios where backends don’t accept bound parameters.

Returns a callable which will receive a literal Python value as the sole positional argument and will return a string representation to be rendered in a SQL statement.

Tip

To provide alternate behavior for _types.TypeEngine.literal_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_literal_param().

See also

Augmenting Existing Types

class calibpipe.database.interfaces.DateTime(timezone: bool = False)[source]#

Bases: _RenderISO8601NoT, HasExpressionLookup, TypeEngine

A type for datetime.datetime() objects.

Date and time types return objects from the Python datetime module. Most DBAPIs have built in support for the datetime module, with the noted exception of SQLite. In the case of SQLite, date and time types are stored as strings which are then converted back to datetime objects when rows are returned.

For the time representation within the datetime type, some backends include additional options, such as timezone support and fractional seconds support. For fractional seconds, use the dialect-specific datatype, such as mysql.TIME. For timezone support, use at least the _types.TIMESTAMP datatype, if not the dialect-specific datatype object.

__init__(timezone: bool = False)[source]#

Construct a new DateTime.

Parameters:: timezone¶ – boolean. Indicates that the datetime type should enable timezone support, if available on the base date/time-holding type only. It is recommended to make use of the _types.TIMESTAMP datatype directly when using this flag, as some databases include separate generic date/time-holding types distinct from the timezone-capable TIMESTAMP datatype, such as Oracle Database.

get_dbapi_type(dbapi)[source]#

Return the corresponding type object from the underlying DB-API, if any.

This can be useful for calling setinputsizes(), for example.

literal_processor(dialect)[source]#

Return a conversion function for processing literal values that are to be rendered directly without using binds.

This function is used when the compiler makes use of the “literal_binds” flag, typically used in DDL generation as well as in certain scenarios where backends don’t accept bound parameters.

Returns a callable which will receive a literal Python value as the sole positional argument and will return a string representation to be rendered in a SQL statement.

Tip

To provide alternate behavior for _types.TypeEngine.literal_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_literal_param().

See also

Augmenting Existing Types

property python_type#

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.

class calibpipe.database.interfaces.NDArray(*args: Any, **kwargs: Any)[source]#

Bases: TypeDecorator

Type for numpy.ndarray binding, include data compression.

The array is stored as a compressed byte string in the database. The class implements the binding between the np.ndarray in the program memory and the byte string stored in the DB.

Compression can be removed or modified, but the two process methods should be the opposite of each other for the binding to work. Ignoring the dialect parameter that is anyway not used, this means that the following assertion should always pass:

db_arr: NDArray
arr: np.ndarray
arr_bytes: bytes = db_arr.process_bind_param(arr)
recov_arr: np.ndarray = db_arr.process_result_value(arr_bytes)
assert(arr == recov_arr)

impl#: alias of LargeBinary

cache_ok: bool = 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

process_bind_param(value: ndarray, dialect) → bytes[source]#

Serialize a np.ndarray into a byte object to store in the DB.

The array is first serialized into bytes and compressed using the default zlib compression algorithm.

process_result_value(value: bytes, dialect) → ndarray[source]#

Deserialize a np.ndarray from bytes read in the DB.

The bytes are first decompressed and the array is loaded from the decompressed byte string.

process_literal_param(value: ndarray, dialect) → str[source]#: Representation of the NDArray object.

property python_type: type#: Return the python type of the underlying object represented by the byte string.

class calibpipe.database.interfaces.SQLTableInfo(table_name: str, metadata: <module 'calibpipe.database.interfaces.sql_metadata' from '/usr/local/lib/python3.12/site-packages/calibpipe/database/interfaces/sql_metadata.py'>, columns: list[~calibpipe.database.interfaces.sql_column_info.SQLColumnInfo], constraints: list[~sqlalchemy.sql.schema.ForeignKeyConstraint | ~sqlalchemy.sql.schema.UniqueConstraint | ~sqlalchemy.sql.schema.CheckConstraint | ~sqlalchemy.sql.schema.PrimaryKeyConstraint] | None = None)[source]#

Bases: object

Collection of attributes defining a Table’s columns.

The class contains the column information (SQLColumnInfo) and additional arguments required to build the sqlalchemy table when the get_table() method is called.

This class can provide useful information on the corresponding table. For example the primary-key or the list of undeferred and deferred columns, i.e. that must be loaded directly or looked up in a cache system (if implemented) respectively. Note that no cache implementation lies here, only the information that some columns must be deferred if possible.

The SQLTableInfo also can manage several tables of the same type (e.g. for versioning, table_A_v1 && table_A_v2). When calling the get_table() method, a custom table name can be given. The object will ensure that only one table is created for a given name (otherwise sqlalchemy cannot work properly).

table_base_class#: alias of Base

__init__(table_name: str, metadata: <module 'calibpipe.database.interfaces.sql_metadata' from '/usr/local/lib/python3.12/site-packages/calibpipe/database/interfaces/sql_metadata.py'>, columns: list[~calibpipe.database.interfaces.sql_column_info.SQLColumnInfo], constraints: list[~sqlalchemy.sql.schema.ForeignKeyConstraint | ~sqlalchemy.sql.schema.UniqueConstraint | ~sqlalchemy.sql.schema.CheckConstraint | ~sqlalchemy.sql.schema.PrimaryKeyConstraint] | None = None) → None[source]#: Initialize the table data and sqlachemy metadata.

get_primary_keys() → list[SQLColumnInfo][source]#

Get list of primary keys for the table.

Returns:

list: list with SQLColumnInfo objects that are the primary keys

Raises:

InvalidTableError: If there are no primary key in the table

get_deferred_columns() → list[SQLColumnInfo][source]#

Return the columns that must be deferred.

Deferred columns won’t be loaded directly when queried.

get_undeferred_columns() → list[SQLColumnInfo][source]#

Return the columns that must not be deferred.

These columns are loaded directly when queried.

get_table(table_name: str | None = None) → Table[source]#

Return a table with a given name, create it if necessary.

Parameters:

table_name: str (optional, default=None): Name of the table to create. If not given, the table_name attribute is used. If the table with the given name has already been created it is returned and no new table is generated.

class calibpipe.database.interfaces.SQLColumnInfo(name: str, field_type: TypeEngine, unit: str | None = '', is_deferred: bool | None = None, **kwargs)[source]#

Bases: object

Contain info required to create a sa.Column object.

The data representing the column is system-independent, using in particular the generic types in .types, only the generate_column() method is specialized for sqlalchemy (returning a sa.Column object).

Attributes:

name: str: Field name
field_type: ColumnType: Field type. See the .types import for possible types
is_deferred: bool (optional, default=None): If given, tell if the field must be deferred i.e. loaded only later (when queried) if a cache system is in place. If not given, only NDArray objects are deferred.

__init__(name: str, field_type: TypeEngine, unit: str | None = '', is_deferred: bool | None = None, **kwargs) → None[source]#

Initialize the column data.

Any keyword argument required to build the final Column object (here sa.Column for sqlachemy) in the generate_column() method can be given to the initializer.

generate_column() → Column[source]#: Generate a new corresponding sa.Column object.

is_primary_key() → bool[source]#: Check if the column is a primary key.

class calibpipe.database.interfaces.HashableRowData(table_name: str, primary_key: Hashable)[source]#

Bases: object

Contain hashable row information (table and primary key).

A table name and a primary key value is enough information to uniquely identify a row inside the entire database. This information can therefore be hashed to create a map indexed by a row.

Attributes:

table_name: str: Name of the table in which the row is stored.
primary_key: Hashable: Any python object that can be hashed, must contain the primary key value of the row.

__init__(table_name: str, primary_key: Hashable) → None[source]#: Initialize hashable table.

table_name: str#

primary_key: Hashable#

class calibpipe.database.interfaces.TableHandler[source]#

Bases: object

Handles tables in CalibPipe DataBase.

The first method returns a valid insertion for a DB, made by the table instance and the values to be inserted. The second method just insert values in a DB, provided the DB connection, the table and the values.

static get_database_table_insertion(container: Container, version: str | None = None) → tuple[Table, dict[str, Any]][source]#: Return a valid insertion for a DB made by the table instance, and the values to insert.

static insert_row_in_database(table: Table, kwargs: dict[str, Any], connection: CalibPipeDatabase) → None[source]#: Insert values in a DB table as a row.

static read_table_from_database(container: Container, connection: CalibPipeDatabase, condition: str | None = None) → QTable[source]#

Read a table from the DB and return it as a QTable object.

An optional argument condition shall have the following form: c.<column_name> <operator> <value> or a combination of thereof using & and | operators. In case of compound condition, every singleton must be contained in parentheses.

static get_compatible_version(version_table: Table, table_name: str, version: str, connection: CalibPipeDatabase) → str[source]#

Get a compatible version for a certain table from the version table.

If no compatible version of the table is available, the new version the table will be added to the version table.

static update_tables_info(table: Table, version_table: Table, table_name: str, comp_version: str, table_version: str, connection: CalibPipeDatabase) → str[source]#

Update the tables’ info.

Updated min and max timestamps are taken from the data table, and a check on version is performed to update the version table. Also, the name of the table is updated accordingly if version has changed.

static update_version_table(version_table: Table, table_name: str, old_version: str, new_version: str, min_time: datetime, max_time: datetime, connection: CalibPipeDatabase) → None[source]#: Update the version of a table with the new version in the version table of the DB.

static update_table_name(table: Table, version: str, connection: CalibPipeDatabase) → None[source]#: Update the name of a table with the new version.

static prepare_db_tables(containers, db_config)[source]#

Create and upload to the CalibPipe DB empty tables for selected calibration containers.

Parameters:

containerslist[Container]: list of calibpipe containers or ContainerMeta instances that will be created as empty tables in the DB
config_datadict: Calibpipe configuration with database connection configuration

static upload_data(calibpipe_data_container: Container, metadata: list[Container] | None, connection: CalibPipeDatabase) → None[source]#

Upload data and optional metadata to the database.

This method uploads the provided data to the main database table and, if provided, associates the metadata with the inserted data row.

Parameters:

calibpipe_data_containerctapipe.core.Container: The data container with the data to be uploaded to the main table of the database.
metadatalist[Container] or None: Optional list of metadata containers to be uploaded. Should include a “ReferenceMetadataContainer” if metadata is provided.
connectionCalibPipeDatabase: An active database connection to the CalibPipe database.

Raises:

DBStorageError: If there are issues with the database connection.
ValueError: If the main table does not contain a single autoincremented primary key or if ReferenceMetadataContainer is missing when metadata is provided.

calibpipe.database.interfaces.query_full_table(table_info: SQLTableInfo, version: str | None = None) → tuple[Any, list[str]][source]#

Return a query for a complete table.

Parameters:

table_info: SQLTableInfo: Table to which the query must be built.
version: Optional[str], default=None: Software version of the data to retrieve. If None is given, the _pro version will be used i.e. the latest available.

Returns:

tuple[Query, list[str]]: A tuple containing the light query to retrieve small fields and the list of field names for deferred fields (cached and loaded later).

calibpipe.database.interfaces.query_from_date(table_info: SQLTableInfo, version: str, date: date) → tuple[Any, list[str]][source]#

Return a query from a date.

Parameters:

table_info: SQLTableInfo: Table to which the query must be built.
version: Optional[str], default=None: Software version of the data to retrieve. If None is given, the _pro version will be used i.e. the latest available.

Returns:

tuple[Query, list[str]]: A tuple containing the light query to retrieve small fields and the list of field names for deferred fields (cached and loaded later).

calibpipe.database.interfaces.query_from_run(table_info: SQLTableInfo, version: str, run: int) → tuple[Any, list[str]][source]#

Return a query from a run.

Parameters:

table_info: SQLTableInfo: Table to which the query must be built.
version: Optional[str], default=None: Software version of the data to retrieve. If None is given, the _pro version will be used i.e. the latest available.

Returns:

tuple[Query, list[str]]: A tuple containing the light query to retrieve small fields and the list of field names for deferred fields (cached and loaded later).

Submodules#

calibpipe.database.interfaces.hashable_row_data module#

HashableRowData class.

class calibpipe.database.interfaces.hashable_row_data.HashableRowData(table_name: str, primary_key: Hashable)[source]#

Bases: object

Contain hashable row information (table and primary key).

A table name and a primary key value is enough information to uniquely identify a row inside the entire database. This information can therefore be hashed to create a map indexed by a row.

Attributes:

table_name: str: Name of the table in which the row is stored.
primary_key: Hashable: Any python object that can be hashed, must contain the primary key value of the row.

__init__(table_name: str, primary_key: Hashable) → None[source]#: Initialize hashable table.

table_name: str#

primary_key: Hashable#

calibpipe.database.interfaces.queries module#

Built-in queries for camera calibration data.

All built-in queries are obtained by calling functions that return a tuple of query and a list of strings, e.g.:

light_query, deferred_column_names = some_builtin_query()

Each time, the query object can be directly sent to an execute() SQLConnection method and it will retrieve from the DB the primary key and all light fields (by default non-array types, see the deferred property of SQLColumnInfo). This can be done e.g. using:

db.execute(light_query)

The list of strings returned in the tuple is the list of deferred fields, that must be queried separately later on, e.g. using:

query = sa.select(deferred_column_names)
res = db.execute(query)

The built-in queries are:

query_full_table(): Return the query utilities to retrieve a full table from the database (used e.g. to retrieve the run metadata table).

query_from_run(): Return the query utilities to retrieve data of a given run in a given table.

query_from_date(): Return the query utilities to retrieve data at a given date in a given table.

The queries have an undefined type (internal to sqlalchemy and not fully defined), an alias to Any is used in this file to express what objects are SQL queries.

calibpipe.database.interfaces.queries.Query#: Alias to express which objects are queries as sqlalchemy does not define it.

calibpipe.database.interfaces.queries.query_full_table(table_info: SQLTableInfo, version: str | None = None) → tuple[Any, list[str]][source]#

Return a query for a complete table.

Parameters:

table_info: SQLTableInfo: Table to which the query must be built.
version: Optional[str], default=None: Software version of the data to retrieve. If None is given, the _pro version will be used i.e. the latest available.

Returns:

tuple[Query, list[str]]: A tuple containing the light query to retrieve small fields and the list of field names for deferred fields (cached and loaded later).

calibpipe.database.interfaces.queries.query_from_date(table_info: SQLTableInfo, version: str, date: date) → tuple[Any, list[str]][source]#

Return a query from a date.

Parameters:

table_info: SQLTableInfo: Table to which the query must be built.
version: Optional[str], default=None: Software version of the data to retrieve. If None is given, the _pro version will be used i.e. the latest available.

Returns:

tuple[Query, list[str]]: A tuple containing the light query to retrieve small fields and the list of field names for deferred fields (cached and loaded later).

calibpipe.database.interfaces.queries.query_from_run(table_info: SQLTableInfo, version: str, run: int) → tuple[Any, list[str]][source]#

Return a query from a run.

Parameters:

table_info: SQLTableInfo: Table to which the query must be built.
version: Optional[str], default=None: Software version of the data to retrieve. If None is given, the _pro version will be used i.e. the latest available.

Returns:

tuple[Query, list[str]]: A tuple containing the light query to retrieve small fields and the list of field names for deferred fields (cached and loaded later).

calibpipe.database.interfaces.sql_column_info module#

SQLColumnInfo class.

class calibpipe.database.interfaces.sql_column_info.SQLColumnInfo(name: str, field_type: TypeEngine, unit: str | None = '', is_deferred: bool | None = None, **kwargs)[source]#

Bases: object

Contain info required to create a sa.Column object.

Attributes:

name: str: Field name
field_type: ColumnType: Field type. See the .types import for possible types
is_deferred: bool (optional, default=None): If given, tell if the field must be deferred i.e. loaded only later (when queried) if a cache system is in place. If not given, only NDArray objects are deferred.

__init__(name: str, field_type: TypeEngine, unit: str | None = '', is_deferred: bool | None = None, **kwargs) → None[source]#

Initialize the column data.

Any keyword argument required to build the final Column object (here sa.Column for sqlachemy) in the generate_column() method can be given to the initializer.

generate_column() → Column[source]#: Generate a new corresponding sa.Column object.

is_primary_key() → bool[source]#: Check if the column is a primary key.

calibpipe.database.interfaces.sql_metadata module#

SQL meta-data variable.

calibpipe.database.interfaces.sql_metadata.sql_metadata = MetaData()#: Metadata variable used for sqlalchemy objects.

calibpipe.database.interfaces.sql_table_info module#

SQLTableInfo class.

exception calibpipe.database.interfaces.sql_table_info.InvalidTableError[source]#

Bases: Exception

Raised when a table is invalid e.g. has no primary key.

class calibpipe.database.interfaces.sql_table_info.SQLTableInfo(table_name: str, metadata: <module 'calibpipe.database.interfaces.sql_metadata' from '/usr/local/lib/python3.12/site-packages/calibpipe/database/interfaces/sql_metadata.py'>, columns: list[~calibpipe.database.interfaces.sql_column_info.SQLColumnInfo], constraints: list[~sqlalchemy.sql.schema.ForeignKeyConstraint | ~sqlalchemy.sql.schema.UniqueConstraint | ~sqlalchemy.sql.schema.CheckConstraint | ~sqlalchemy.sql.schema.PrimaryKeyConstraint] | None = None)[source]#

Bases: object

Collection of attributes defining a Table’s columns.

The class contains the column information (SQLColumnInfo) and additional arguments required to build the sqlalchemy table when the get_table() method is called.

table_base_class#: alias of Base

__init__(table_name: str, metadata: <module 'calibpipe.database.interfaces.sql_metadata' from '/usr/local/lib/python3.12/site-packages/calibpipe/database/interfaces/sql_metadata.py'>, columns: list[~calibpipe.database.interfaces.sql_column_info.SQLColumnInfo], constraints: list[~sqlalchemy.sql.schema.ForeignKeyConstraint | ~sqlalchemy.sql.schema.UniqueConstraint | ~sqlalchemy.sql.schema.CheckConstraint | ~sqlalchemy.sql.schema.PrimaryKeyConstraint] | None = None) → None[source]#: Initialize the table data and sqlachemy metadata.

get_primary_keys() → list[SQLColumnInfo][source]#

Get list of primary keys for the table.

Returns:

list: list with SQLColumnInfo objects that are the primary keys

Raises:

InvalidTableError: If there are no primary key in the table

get_deferred_columns() → list[SQLColumnInfo][source]#

Return the columns that must be deferred.

Deferred columns won’t be loaded directly when queried.

get_undeferred_columns() → list[SQLColumnInfo][source]#

Return the columns that must not be deferred.

These columns are loaded directly when queried.

get_table(table_name: str | None = None) → Table[source]#

Return a table with a given name, create it if necessary.

Parameters:

table_name: str (optional, default=None): Name of the table to create. If not given, the table_name attribute is used. If the table with the given name has already been created it is returned and no new table is generated.

calibpipe.database.interfaces.table_handler module#

Utilities for CalibPipe data.

class calibpipe.database.interfaces.table_handler.TableHandler[source]#

Bases: object

Handles tables in CalibPipe DataBase.

static get_database_table_insertion(container: Container, version: str | None = None) → tuple[Table, dict[str, Any]][source]#: Return a valid insertion for a DB made by the table instance, and the values to insert.

static insert_row_in_database(table: Table, kwargs: dict[str, Any], connection: CalibPipeDatabase) → None[source]#: Insert values in a DB table as a row.

static read_table_from_database(container: Container, connection: CalibPipeDatabase, condition: str | None = None) → QTable[source]#

Read a table from the DB and return it as a QTable object.

static get_compatible_version(version_table: Table, table_name: str, version: str, connection: CalibPipeDatabase) → str[source]#

Get a compatible version for a certain table from the version table.

If no compatible version of the table is available, the new version the table will be added to the version table.

static update_tables_info(table: Table, version_table: Table, table_name: str, comp_version: str, table_version: str, connection: CalibPipeDatabase) → str[source]#

Update the tables’ info.

static update_version_table(version_table: Table, table_name: str, old_version: str, new_version: str, min_time: datetime, max_time: datetime, connection: CalibPipeDatabase) → None[source]#: Update the version of a table with the new version in the version table of the DB.

static update_table_name(table: Table, version: str, connection: CalibPipeDatabase) → None[source]#: Update the name of a table with the new version.

static prepare_db_tables(containers, db_config)[source]#

Create and upload to the CalibPipe DB empty tables for selected calibration containers.

Parameters:

containerslist[Container]: list of calibpipe containers or ContainerMeta instances that will be created as empty tables in the DB
config_datadict: Calibpipe configuration with database connection configuration

static upload_data(calibpipe_data_container: Container, metadata: list[Container] | None, connection: CalibPipeDatabase) → None[source]#

Upload data and optional metadata to the database.

This method uploads the provided data to the main database table and, if provided, associates the metadata with the inserted data row.

Parameters:

calibpipe_data_containerctapipe.core.Container: The data container with the data to be uploaded to the main table of the database.
metadatalist[Container] or None: Optional list of metadata containers to be uploaded. Should include a “ReferenceMetadataContainer” if metadata is provided.
connectionCalibPipeDatabase: An active database connection to the CalibPipe database.

Raises:

DBStorageError: If there are issues with the database connection.
ValueError: If the main table does not contain a single autoincremented primary key or if ReferenceMetadataContainer is missing when metadata is provided.

calibpipe.database.interfaces.types module#

Type definitions for SQLAlchemy.

These type definitions allow us to define database fields and containers being almost completely decoupled from SQLAlchemy (without direct coupling).

In particular, SQLColumnInfo and SQLTableInfo use these generic types and not the sqlalchemy types directly.

The NDArray type is defined explicitly to implemented the serialization/deserialization np.ndarray <-> bytes and the (optional) zlib compression/decompression on the byte data.

class calibpipe.database.interfaces.types.NDArray(*args: Any, **kwargs: Any)[source]#

Bases: TypeDecorator

Type for numpy.ndarray binding, include data compression.

The array is stored as a compressed byte string in the database. The class implements the binding between the np.ndarray in the program memory and the byte string stored in the DB.

db_arr: NDArray
arr: np.ndarray
arr_bytes: bytes = db_arr.process_bind_param(arr)
recov_arr: np.ndarray = db_arr.process_result_value(arr_bytes)
assert(arr == recov_arr)

impl#: alias of LargeBinary

cache_ok: bool = True#

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

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

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

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

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

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

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    this is the non-cacheable version, as "self.lookup" is not
    hashable.

    """

    def __init__(self, lookup):
        self.lookup = lookup

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self.lookup" ...

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

>>> type_ = LookupType({"a": 10, "b": 20})
>>> type_._static_cache_key
<stdin>:1: SAWarning: UserDefinedType LookupType({'a': 10, 'b': 20}) will not
produce a cache key because the ``cache_ok`` flag is not set to True.
Set this flag to True if this type object's state is safe to use
in a cache key, or False to disable this warning.
symbol('no_cache')

>>> # set cache_ok = True
>>> type_.cache_ok = True

>>> # this is the cache key it would generate
>>> key = type_._static_cache_key
>>> key
(<class '__main__.LookupType'>, ('lookup', {'a': 10, 'b': 20}))

>>> # however this key is not hashable, will fail when used with
>>> # SQLAlchemy statement cache
>>> some_cache = {key: "some sql value"}
Traceback (most recent call last): File "<stdin>", line 1,
in <module> TypeError: unhashable type: 'dict'

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

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    The dictionary is stored both as itself in a private variable,
    and published in a public variable as a sorted tuple of tuples,
    which is hashable and will also return the same value for any
    two equivalent dictionaries.  Note it assumes the keys and
    values of the dictionary are themselves hashable.

    """

    cache_ok = True

    def __init__(self, lookup):
        self._lookup = lookup

        # assume keys/values of "lookup" are hashable; otherwise
        # they would also need to be converted in some way here
        self.lookup = tuple((key, lookup[key]) for key in sorted(lookup))

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self._lookup" ...

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

>>> LookupType({"a": 10, "b": 20})._static_cache_key
(<class '__main__.LookupType'>, ('lookup', (('a', 10), ('b', 20))))

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

process_bind_param(value: ndarray, dialect) → bytes[source]#

Serialize a np.ndarray into a byte object to store in the DB.

The array is first serialized into bytes and compressed using the default zlib compression algorithm.

process_result_value(value: bytes, dialect) → ndarray[source]#

Deserialize a np.ndarray from bytes read in the DB.

The bytes are first decompressed and the array is loaded from the decompressed byte string.

process_literal_param(value: ndarray, dialect) → str[source]#: Representation of the NDArray object.

property python_type: type#: Return the python type of the underlying object represented by the byte string.

calibpipe.database.interfaces package#

Submodules#

calibpipe.database.interfaces.hashable_row_data module#

calibpipe.database.interfaces.queries module#

calibpipe.database.interfaces.sql_column_info module#

calibpipe.database.interfaces.sql_metadata module#

calibpipe.database.interfaces.sql_table_info module#

calibpipe.database.interfaces.table_handler module#

calibpipe.database.interfaces.types module#

This Page