Skip to content

Service

LITERAL_STATUS = Literal['exists', 'created', 'updated'] module-attribute

logger = logging.getLogger(__name__) module-attribute

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

SessionDatabaseConfigUnitOfWork

Bases: SessionDatabaseUnitOfWork, AbstractConfigUnitOfWork

repository_class = SQLConfigRepository class-attribute

build_default_assistant_schemas()

Build the default assistant schemas.

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

load_text(path)

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.