repository

common

class EntityDataProcessor(session: AsyncSession, all_data: SQLAlchemyRepository, selector: EntityDataSelector)

Bases: object

Logic for storing and retrieving entity data from the database. This can be used for both Dataset data and Update data

async get(id: UUID) → Dict[str, Dict[str, UniformAttributeData | CSRAttributeData]]
async store(id: UUID, data: Dict[str, Dict[str, UniformAttributeData | CSRAttributeData]])
Parameters:

  • id – dataset UUID

  • data – dataset data section in numpy format

class EntityDataSelector(*args, **kwargs)

Bases: Protocol

EntityDataSelector is a Protocol that EntityDataProcessor uses to retrieve SQLAlchemy queries for the resource it is operating on, ie Dataset or Update.

insert_linked_attribute(id: UUID, attribute_id: UUID) → sqlalchemy.Insert
select_linked_attribute(id: UUID) → sqlalchemy.Select.tuple[movici_data_core.database.model.Attribute]
class GenericResourceRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository)

Bases: SQLResourceRepository, Generic[T_dom]

A GenericResourceRepository is the simplest CRUD repository. Resources are globally unique by name and do not have a parent (such as a Workspace). Resources that are managed through a GenericResourceRepository are:

  • Workspaces

  • DatasetTypes

  • EntityTypes

  • AttributeTypes

  • ModelTypes

To implement a GenericResourceRepository, subclass it and set the __resource__ and __resource_type_name__ class fields. Then, implement the create and update methods it is recommended to wrap the update method in an @ensure_valid_id decorator

async create(obj: T_dom) → UUID
async delete(id: UUID)
async exists(name: str) → bool
async get_by_id(id: UUID) → T_dom | None
async get_by_name(name: str) → T_dom | None
async list() → list[T_dom]
async update(id: UUID, obj: T_dom) → None
class RawDataProcessor(session: sqlalchemy.ext.asyncio.AsyncSession)

Bases: object

Logic for storing dataset data as raw bytes in the database. Used for BINARY and UNSTRUCTURED datasets

RAW_DATA_CHUNK_SIZE = 100000000
async get(id: UUID | None = None, raw_data: RawData | None = None) → tuple[str | None, bytearray]
async get_dict(id: UUID) → dict
async store(id: UUID, data: dict | bytes | BinaryIO | Path, chunk_size=0)
async stream_bytes(id: UUID | None = None, raw_data: RawData | None = None, yield_per=1) → tuple[str | None, AsyncGenerator[bytes, None]]
Returns:

a tuple (encoding, bytestreamer). The bytestreamer can be used as an async generator

class SQLResourceRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository)

Bases: object

Base class for the various resource repositories

all_data: SQLAlchemyRepository
options: Options
session: AsyncSession
ensure_valid_id(method)

Decorator to ensure that a resource id exists, raises ResourceDoesNotExist if a non-existing id is given :param method: a method from a GenericResourceRepository that takes in id as its first argument

dataset

class DatasetDataRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository)

Bases: SQLResourceRepository

async create(id: UUID, data: dict | bytes | BinaryIO | Path, format: DatasetFormat, chunk_size=0)

Store dataset data for a dataset. The dataset must currently not contain any data

Parameters:

  • id – A dataset id

  • data – The dataset data as dict, bytes, BytesIO or pathlib.Path

  • format – The dataset’s DatasetFormat

  • chunk_size – The maximum chunk size in bytes to store data. By default set to the value of DatasetRepository.RAW_DATA_CHUNK_SIZE. This parameter is ignore when the dataset format is DatasetFormat.ENTITY_BASED

async delete(id: UUID, prune_dataset=True)
async exists_for(id: UUID)
Parameters:

id – a dataset id

Returns:

a boolean wether the dataset has data (either entity data or raw data)

async get_entity_data(id: UUID)

return the dataset data for an ENTITY_BASED dataset

Parameters:

id – the dataset UUID

Returns:

The entity based dataset data section

async get_unstructured_data(id: UUID) → dict

return the dataset data for an UNSTRUCTURED dataset

Parameters:

id – the dataset UUID

Returns:

The unstructured dataset data section

async stream_binary_data(id: UUID, yield_per=1) → AsyncGenerator[bytes, None]
Parameters:

  • id – the dataset UUID

  • yield_per – an optimzation parameter for the database to request n chunks at a time. Default: 1

Returns:

An async generator that can be iterated over (async for loop) to produce chunks of binary data. The chunk size is determined by the data stored in the database and is RawDataProcessor.RAW_DATA_CHUNK_SIZE by default

class DatasetDataSelector(*args, **kwargs)

Bases: EntityDataSelector

insert_linked_attribute(id: UUID, attribute_id: UUID) → sqlalchemy.Insert
select_linked_attribute(id: UUID) → sqlalchemy.Select.tuple[movici_data_core.database.model.Attribute]
class DatasetRepository(session: 'AsyncSession', options: 'Options', all_data: 'SQLAlchemyRepository', workspace_id: 'UUID | None' = None)

Bases: SQLResourceRepository

async create(obj: Dataset) → UUID
async delete(id: UUID)
async ensure_scenario_datasets(datasets: Sequence[ScenarioDataset]) → List[ScenarioDataset]

Ensure that the Datasets of a sequences of ScenarioDatasets exist in the database or raise an error. If one or more of the Dataset``s do not exist and the database option ``STRICT_SCENARIO_DATASETS is unset, the non-existing datasets will be created as a stub. If any of the newly created datasets has a dataset type that does not yet exist, then the STRICT_DATASET_TYPES database option must also be unset. Otherwise an error will be raise. An error is also raised if a Dataset already exists but with a different dataset type.

Parameters:

datasets – The ScenarioDatasets to ensure

Returns:

The datasets as they exist in the database, as ScenarioDataset objects, in the same order as the input sequence

async exists(name: str)

return whether the dataset (by name) already exists in the database for the acitve workspace.

Parameters:

name – dataset name

Returns:

a boolean indicating the dataset exists for the active workspace

async get_by_id(id: UUID) → Dataset | None
async get_by_name(name: str) → Dataset | None
async get_summary(id: UUID)

Request a DatasetSummary for the dataset id

Parameters:

id – the dataset id

Returns:

A DatasetSummary

async list() → list[Dataset]
property selector
property selector_with_has_data
async update(id: UUID, obj: Dataset, chunk_size=0)

Update a dataset. When not given obj.data, only the name and display_name are updated. Otherwise this method als processes the data and updates the general, epsg_code and bounding_box field.

Parameters:

  • id – the UUID of the stored Dataset

  • obj – the Dataset object with the changes

  • chunk_size – the chunk size in bytes when storing unstructured or binary data. When set to 0, the default, the chunk size is set to RawDataProcessor.RAW_DATA_CHUNK_SIZE

workspace_id: UUID | None = None

general

class AttributeTypeRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository)

Bases: GenericResourceRepository[AttributeType]

async create(obj: AttributeType) → UUID

Store a :class:AttributeType in the database

Parameters:

obj – the AttributeType object

Returns:

the UUID of the stored AttributeType

async ensure_attribute_type(attribute_type: AttributeType) → AttributeType

Ensure that an attribute type exists in the database or raise an error. If the attribute type does not exist and the database option STRICT_ATTRIBUTE_TYPES is unset, the attribute type will be created. If the STRICT_ATTRIBUTE_TYPES options is set, an error is raised instead. An error will also be raised if an AttributeType with a different data_type already exists.

Parameters:

attribute_type – the AttributeType object to ensure.

Returns:

the AttributeType object as it exists in the database

async update(id: UUID, obj: AttributeType)

Update a :class:AttributeType in the database

Valid fields to update are: name, data_type, unit, description, enum_name

Parameters:

  • id – the UUID of the stored EntityType

  • obj – the EntityType object with the changes

class DatasetTypeRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository)

Bases: GenericResourceRepository[DatasetType]

async create(obj: DatasetType) → UUID

Store a :class:DatasetType in the database. When storing a DatasetType, its format field may not be set to None.

Parameters:

obj – the DatasetType object

Returns:

the UUID of the stored DatasetType

async ensure_dataset_type(dataset_type: DatasetType) → DatasetType

Ensure that a dataset type exists in the database or raise an error. If the dataset type does not exist and the database option STRICT_DATASET_TYPES is unset, the dataset type will be created. If the STRICT_DATASET_TYPES options is set, an error is raised instead. An error will also be raised if an attempt is made to create a DatasetType with its format field set to None

Parameters:

dataset_type – the DatasetType object to ensure.

Returns:

the DatasetType object as it exists in the database

async update(id: UUID, obj: DatasetType)

Update a :class:DatasetType in the database

Valid fields to update are: name, mimetype

Parameters:

  • id – the UUID of the stored DatasetType

  • obj – the DatasetType object

class EntityTypeRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository)

Bases: GenericResourceRepository[EntityType]

async create(obj: EntityType) → UUID

Store a :class:EntityType in the database

Parameters:

obj – the EntityType object

Returns:

the UUID of the stored EntityType

async ensure_entity_type(entity_type: EntityType) → EntityType
async update(id: UUID, obj: EntityType)

Update a :class:EntityType in the database

Valid fields to update are: name

Parameters:

  • id – the UUID of the stored EntityType

  • obj – the EntityType object with the changes

class ModelTypeRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository)

Bases: GenericResourceRepository[ModelType]

async create(obj: ModelType) → UUID

Store a :class:ModelType in the database

Parameters:

obj – the ModelType object

Returns:

the UUID of the stored ModelType

async ensure_model_types(model_types: Sequence[ModelType]) → list[ModelType]

Ensure that a sequence of model types exist in the database or raise an error. If one or more of the model types does not exist and the database option STRICT_MODEL_TYPES is unset, the non-existing model types will be created. If the STRICT_MODEL_TYPES options is set, an error is raised instead.

Parameters:

model_types – The model types to ensure, as a sequence of model type names

Returns:

the ModelType objects as they exist in the database, in the same order as the input sequence

async update(id: UUID, obj: ModelType)

Update a :class:ModelType in the database

Valid fields to update are: name, jsonschema

Parameters:

  • id – the UUID of the stored ModelType

  • obj – the ModelType object with the changes

scenario

class ScenarioRepository(session: 'AsyncSession', options: 'Options', all_data: 'SQLAlchemyRepository', workspace_id: 'UUID | None' = None, scenario_id: 'UUID | None' = None)

Bases: SQLResourceRepository

async create(obj: Scenario, validator: ModelConfigValidator) → UUID

Store a new Scenario in the database. This method can only be invoked if there is no currently active Scenario, to prevent creating scenarios when in a SINGLE_SCENARIO mode

Parameters:

  • obj – The scenario to create

  • validator – A ModelConfigValidator that is instantiated with all available models in the dataset

Returns:

the newly created Scenario UUID

async delete()

Delete de active scenario, if it exists

async exists()
async exists_by_name(name: str)

checks whether a scenario with a specific name exists in the active workspace :return: bool

for_id(scenario_id: UUID)

Bind the ScenarioRepository to a specific scenario

async get() → Scenario | None

Get the active scenario from the database

Returns:

The Scenario, or None if it does not exist

async get_by_name(name: str) → Scenario | None

Get a scenario by name, in the active workspace

async list() → list[Scenario]

List all scenarios in the active workspace

scenario_id: UUID | None = None
property selector
async set_status(status: ScenarioStatus)

Set the ScenarioStatus for the active scenario

async update(obj: Scenario, validator: ModelConfigValidator)

Update a scenario in the database. The scenario to update must be the active scenario

Parameters:

  • obj – The sc

  • validator – A ModelConfigValidator that is instantiated with all available models in the dataset

workspace_id: UUID | None = None

updates

class UpdateDataSelector(*args, **kwargs)

Bases: EntityDataSelector

Subclass of EntityDataSelector to be use for Updates

insert_linked_attribute(id: UUID, attribute_id: UUID)
select_linked_attribute(id: UUID)
class UpdateRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository, scenario_id: UUID | None)

Bases: SQLResourceRepository

A Repository for managing Updates. In contrast to other resources, Updates are immutable and can only be created (added to a scenario). The only way to delete an update is by deleting all updates for a Scenario, thereby effectively resetting the Scenario

Parameters:

scenario_id – A Scenario UUID to bind this UpdateRepository to. Most methods require the repository to be bound, with the exception of UpdateRepository.get_by_id(). Binding generally is performed by the SQLAlchemyRepository that manages this UpdateRepository

async create(obj: Update) → UUID

Store an Update to the active scenario.

Parameters:

obj – the Update to add, including data

Returns:

the newly created Update UUID

async delete_all()

Delete all updates for the active scenario. Should be combined with a separate call, to the Scenario, resetting the ScenarioStatus

async exists() → bool

Does the active scenario have any updates?

async get_by_id(id: UUID, with_data=False) → Update | None
async list() → list[Update]

List all updates in the active scenario.

Returns:

a list of Updates, these Updates do not contain any data

scenario_id: UUID | None
property selector

workspace

class WorkspaceRepository(session: AsyncSession, options: Options, all_data: SQLAlchemyRepository)

Bases: GenericResourceRepository[Workspace]

async create(obj: Workspace) → UUID

Store a :class:Workspace in the database

Parameters:

obj – the Workspace object

Returns:

the UUID of the stored ModelType

async list() → list[Workspace]
async update(id: UUID, obj: Workspace)

Update a :class:Workspace in the database

Valid fields to update are: name, display_name

Parameters:

  • id – the UUID of the stored Workspace

  • obj – the Workspace object with the changes

async with_counts(workspace: Workspace) → Workspace

add Workspace.dataset_count and Workspace.scenario_count to the workspace. The workspace must have been previously loaded from the database