init

__all__ = ['AbstractConfigRepository', 'AbstractConfigUnitOfWork', 'SQLConfigRepository', 'SessionDatabaseConfigUnitOfWork', 'SetDefaultConfigsError', 'TypeNotSetError', 'create_new', 'get', 'initialize_default_configs', 'list_sub_assistant_uids', 'list_uids', 'mark_deleted', 'recursive_ensure_schema_exists', 'set_schema', 'update'] module-attribute

AbstractConfigRepository

Bases: RepositoryBase[ConfigSchemaBase, ConfigID], ABC

create(schema, foreign_keys=None) async

get(unique_id, *, include_deleted=False, fully_load=True, expected_schema_class=None, **kwargs) async

get(
    unique_id: ConfigID,
    *,
    include_deleted: bool = ...,
    fully_load: bool = ...,
    expected_schema_class: None = None,
) -> ConfigSchemaBase

get(
    unique_id: ConfigID,
    *,
    include_deleted: bool = ...,
    fully_load: bool = ...,
    expected_schema_class: type[ConfigSchemaType],
) -> ConfigSchemaType

Load a schema from the database by name and type.

PARAMETER	DESCRIPTION
unique_id	The unique id of the schema to load TYPE: ConfigID
include_deleted	Whether to include deleted schemas in the search TYPE: bool DEFAULT: False
fully_load	Whether to fully load the schema (e.g. load all sub-schemas) TYPE: bool DEFAULT: True
expected_schema_class	The expected schema class to check against and for type hinting TYPE: type[ConfigSchemaType] | None DEFAULT: None
kwargs	Additional keyword arguments DEFAULT: {}

RAISES	DESCRIPTION
NotFoundError	If the schema does not exist
DeletedError(NotFoundError)	If schema is found but marked deleted

list_sub_assistant_uids(type_, user_id='public') async

List the unique ids of all assistant schemas of a given type (and user).

list_uids(schema_class, type_, user_id=None) async

List the unique ids of all schemas of a given type (and user).

mark_deleted(unique_id, nested=False) async

Mark a schema as deleted in the database.

This should act as effectively deleting, but without actually removing from the database. Just setting a flag instead. This should prevent most operations from returning the schema.

set(schema) async

Set a schema in the database.

If the schema already exists, it will be updated. Otherwise, it will be created.

update(schema, foreign_keys=None, include_deleted=False, fully_load=False) async

AbstractConfigUnitOfWork

Bases: AbstractDatabaseUnitOfWork, ABC

Abstract class for a unit of work involving configuration data.

Any modifications/creations/deletions of configuration data should be done within a unit of work.

repository instance-attribute

flush() async

SQLConfigRepository

Bases: SQLRepositoryMixin[ConfigID], AbstractConfigRepository

UID = ConfigID class-attribute instance-attribute

__check_type(schema) staticmethod

SessionDatabaseConfigUnitOfWork

Bases: SessionDatabaseUnitOfWork, AbstractConfigUnitOfWork

repository_class = SQLConfigRepository class-attribute

SetDefaultConfigsError

Bases: BackendError

Error setting default configs.

TypeNotSetError

Bases: BackendError

__init__()

create_new(uow, schema) async

Create a completely new configuration schema.

I.e., one that has not been previously saved in the database and does not already exist in the database.

get(uow, unique_id, schema_class=None) async

get(
    uow: AbstractConfigUnitOfWork,
    unique_id: ConfigID,
    schema_class: type[ConfigSchemaBaseType],
) -> ConfigSchemaBaseType

get(
    uow: AbstractConfigUnitOfWork,
    unique_id: ConfigID,
    schema_class: None = ...,
) -> ConfigSchemaBase

Get a schema from the database by its unique id.

PARAMETER	DESCRIPTION
uow	Unit of work to use TYPE: AbstractConfigUnitOfWork
unique_id	Unique id of the schema to get TYPE: ConfigID
schema_class	Optional expected return schema class (for validation and type checking) TYPE: type[ConfigSchemaBaseType] | None DEFAULT: None

RETURNS	DESCRIPTION
ConfigSchemaBaseType	The schema from the database

initialize_default_configs(uow) async

Ensure that the default configs are in the database.

list_sub_assistant_uids(uow, type_, user_id='public') async

list_uids(uow, schema_class, type_, user_id=None) async

mark_deleted(uow, unique_id) async

Mark a schema as deleted in the database.

This should act as effectively deleting, but without actually removing from the database. Just setting a flag instead. This should prevent most operations from returning the schema.

recursive_ensure_schema_exists(uow, schema) async

Ensure that schema and all sub-schemas (recursively) exist in the database.

This avoids trying to create a schema that already exists, but also makes sure that any nested schemas that do not exist are created first.

2024-11-05 -- Mostly intended for testing purposes where whole configs are created in one go.

set_schema(uow, schema) async

Set a schema in the database.

If the schema already exists, it will be updated. Otherwise, it will be created.

update(uow, schema) async

Update an existing schema in the database.

This will create a new entry in the database for the UniqueID with an incremented version number.