lomas_server.admin_database package

Submodules

lomas_server.admin_database.admin_database module

class lomas_server.admin_database.admin_database.AdminDatabase

Bases: ABC, BaseModel

Overall database management for server state.

abstract does_dataset_exist(dataset_name: str) → bool

Checks if dataset exist in the database.

Parameters:

dataset_name (str) – name of the dataset to check

Returns:

True if the dataset exists, False otherwise.

Return type:

bool

abstract does_user_exist(user_name: str) → bool

Checks if user exist in the database.

Parameters:

user_name (str) – name of the user to check

Returns:

True if the user exists, False otherwise.

Return type:

bool

abstract get_and_set_may_user_query(user_name: str, may_query: bool) → bool

Atomic operation to check and set if the user may query the server.

(Set False before querying and True after updating budget)

Wrapped by user_must_exist().

Parameters:

• user_name (str) – name of the user

• may_query (bool) – flag give or remove access to user

Returns:

The may_query status of the user before the update.

Return type:

bool

abstract get_dataset(dataset_name: str) → DSInfo

Get dataset access info based on dataset_name.

Wrapped by dataset_must_exist().

Parameters:

dataset_name (str) – Name of the dataset.

Returns:

The dataset model.

Return type:

Dataset

abstract get_dataset_metadata(dataset_name: str) → Metadata

Returns the metadata dictionnary of the dataset.

Wrapped by dataset_must_exist().

Parameters:

dataset_name (str) – name of the dataset to get the metadata

Returns:

The metadata object.

Return type:

Metadata

abstract get_epsilon_or_delta(user_name: str, dataset_name: str, parameter: BudgetDBKey) → float

Get the total spent epsilon or delta by user on dataset.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• parameter (str) – Member of BudgetDBKey.

Returns:

The requested budget value.

Return type:

float

get_initial_budget(user_name: str, dataset_name: str) → list[float]

Get the initial epsilon and delta budget.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

The first value of the list is the epsilon value,

the second value is the delta value.

Return type:

List[float]

get_remaining_budget(user_name: str, dataset_name: str) → list[float]

Get the remaining epsilon and delta budget (initial - total spent).

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

The first value of the list is the epsilon value,

the second value is the delta value.

Return type:

List[float]

get_total_spent_budget(user_name: str, dataset_name: str) → list[float]

Get the total spent epsilon and delta spent by user on dataset.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

The first value of the list is the epsilon value,

the second value is the delta value.

Return type:

List[float]

abstract get_user_previous_queries(user_name: str, dataset_name: str) → list[dict]

Retrieves and return the queries already done by a user.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

List of previous queries.

Return type:

List[dict]

abstract has_user_access_to_dataset(user_name: str, dataset_name: str) → bool

Checks if a user may access a particular dataset.

Wrapped by user_must_exist().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

True if the user has access, False otherwise.

Return type:

bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prepare_save_query(user_name: str, query: LomasRequestModel, response: QueryResponse) → dict

Prepare the query to save in archives.

Parameters:

• user_name (str) – name of the user

• query (LomasRequestModel) – Request object received from client

• response (QueryResponse) – Response object sent to client

Raises:

InternalServerException – If the type of query is unknown.

Returns:

The query archive dictionary.

Return type:

dict

abstract save_query(user_name: str, query: LomasRequestModel, response: QueryResponse) → None

Save queries of user on datasets in a separate collection (table).

Parameters:

• user_name (str) – name of the user

• query (LomasRequestModel) – Request object received from client

• response (QueryResponse) – Response object sent to client

set_may_user_query(user_name: str, may_query: bool) → None

Sets if a user may query the server..

(Set False before querying and True after updating budget)

Wrapped by user_must_exist().

Parameters:

• user_name (str) – name of the user

• may_query (bool) – flag give or remove access to user

update_budget(user_name: str, dataset_name: str, spent_epsilon: float, spent_delta: float) → None

Update current epsilon and delta delta spent by user.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• spent_epsilon (float) – value of epsilon spent on last query

• spent_delta (float) – value of delta spent on last query

update_delta(user_name: str, dataset_name: str, spent_delta: float) → None

Update spent delta spent by user with spent delta of the user.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• spent_delta (float) – value of delta spent on last query

update_epsilon(user_name: str, dataset_name: str, spent_epsilon: float) → None

Update spent epsilon by user with total spent epsilon.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• spent_epsilon (float) – value of epsilon spent on last query

abstract update_epsilon_or_delta(user_name: str, dataset_name: str, parameter: BudgetDBKey, spent_value: float) → None

Update current budget spent by user with spent budget.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• parameter (str) – One of BudgetDBKey

• spent_value (float) – spending of epsilon or delta on last query

abstract wipe() → None

Wipe the entire Database.

lomas_server.admin_database.admin_database.dataset_must_exist(func: Callable[[Concatenate[DB, str, P]], T]) → Callable[[Concatenate[DB, str, P]], T]

Decorator function to verify that a dataset exists.

Parameters:

func (Callable) – Function to be decorated. Wrapped function arguments must include: - args[0] (str): dataset name

Raises:

InvalidQueryException – If the dataset does not exist.

Returns:

Wrapper function that checks if the dataset exists

before calling the wrapped function.

Return type:

Callable

lomas_server.admin_database.admin_database.user_must_exist(func: Callable[[Concatenate[DB, str, P]], T]) → Callable[[Concatenate[DB, str, P]], T]

Decorator function to verify that a user exists.

Parameters:

func (Callable) – Function to be decorated. Wrapped function arguments must include: - args[0] (str): username

Raises:

UnauthorizedAccessException – If the user does not exist.

Returns:

Wrapper function that verifies the user exists

before calling func.

Return type:

Callable

lomas_server.admin_database.admin_database.user_must_have_access_to_dataset(func: Callable[[Concatenate[DB, str, str, P]], T]) → Callable[[Concatenate[DB, str, str, P]], T]

Decorator function to enforce a user has access to a dataset.

Parameters:

func (Callable) – Function to be decorated. Wrapped function arguments must include: - args[0] (str): user name - args[1] (str): dataset name

Raises:

UnauthorizedAccessException – If the user does not have access to the dataset.

Returns:

Wrapper function that checks if the user has access

to the dataset before calling the wrapped function.

Return type:

Callable

lomas_server.admin_database.constants module

class lomas_server.admin_database.constants.BudgetDBKey(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Key for selecting budget values in admin db for given.

dataset and user.

DELTA_INIT = 'initial_delta'

DELTA_SPENT = 'total_spent_delta'

EPSILON_INIT = 'initial_epsilon'

EPSILON_SPENT = 'total_spent_epsilon'

class lomas_server.admin_database.constants.TopDBKey(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Key of the top level collecions.

ARCHIVE = 'queries_archive'

DATASETS = 'datasets'

METADATA = 'metadata'

USERS = 'users'

lomas_server.admin_database.factory module

lomas_server.admin_database.mongodb_database module

lomas_server.admin_database.utils module

lomas_server.admin_database.yaml_database module

Module contents

class lomas_server.admin_database.AdminDatabase

Bases: ABC, BaseModel

Overall database management for server state.

abstract does_dataset_exist(dataset_name: str) → bool

Checks if dataset exist in the database.

Parameters:

dataset_name (str) – name of the dataset to check

Returns:

True if the dataset exists, False otherwise.

Return type:

bool

abstract does_user_exist(user_name: str) → bool

Checks if user exist in the database.

Parameters:

user_name (str) – name of the user to check

Returns:

True if the user exists, False otherwise.

Return type:

bool

abstract get_and_set_may_user_query(user_name: str, may_query: bool) → bool

Atomic operation to check and set if the user may query the server.

(Set False before querying and True after updating budget)

Wrapped by user_must_exist().

Parameters:

• user_name (str) – name of the user

• may_query (bool) – flag give or remove access to user

Returns:

The may_query status of the user before the update.

Return type:

bool

abstract get_dataset(dataset_name: str) → DSInfo

Get dataset access info based on dataset_name.

Wrapped by dataset_must_exist().

Parameters:

dataset_name (str) – Name of the dataset.

Returns:

The dataset model.

Return type:

Dataset

abstract get_dataset_metadata(dataset_name: str) → Metadata

Returns the metadata dictionnary of the dataset.

Wrapped by dataset_must_exist().

Parameters:

dataset_name (str) – name of the dataset to get the metadata

Returns:

The metadata object.

Return type:

Metadata

abstract get_epsilon_or_delta(user_name: str, dataset_name: str, parameter: BudgetDBKey) → float

Get the total spent epsilon or delta by user on dataset.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• parameter (str) – Member of BudgetDBKey.

Returns:

The requested budget value.

Return type:

float

get_initial_budget(user_name: str, dataset_name: str) → list[float]

Get the initial epsilon and delta budget.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

The first value of the list is the epsilon value,

the second value is the delta value.

Return type:

List[float]

get_remaining_budget(user_name: str, dataset_name: str) → list[float]

Get the remaining epsilon and delta budget (initial - total spent).

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

The first value of the list is the epsilon value,

the second value is the delta value.

Return type:

List[float]

get_total_spent_budget(user_name: str, dataset_name: str) → list[float]

Get the total spent epsilon and delta spent by user on dataset.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

The first value of the list is the epsilon value,

the second value is the delta value.

Return type:

List[float]

abstract get_user_previous_queries(user_name: str, dataset_name: str) → list[dict]

Retrieves and return the queries already done by a user.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

List of previous queries.

Return type:

List[dict]

abstract has_user_access_to_dataset(user_name: str, dataset_name: str) → bool

Checks if a user may access a particular dataset.

Wrapped by user_must_exist().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

True if the user has access, False otherwise.

Return type:

bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prepare_save_query(user_name: str, query: LomasRequestModel, response: QueryResponse) → dict

Prepare the query to save in archives.

Parameters:

• user_name (str) – name of the user

• query (LomasRequestModel) – Request object received from client

• response (QueryResponse) – Response object sent to client

Raises:

InternalServerException – If the type of query is unknown.

Returns:

The query archive dictionary.

Return type:

dict

abstract save_query(user_name: str, query: LomasRequestModel, response: QueryResponse) → None

Save queries of user on datasets in a separate collection (table).

Parameters:

• user_name (str) – name of the user

• query (LomasRequestModel) – Request object received from client

• response (QueryResponse) – Response object sent to client

set_may_user_query(user_name: str, may_query: bool) → None

Sets if a user may query the server..

(Set False before querying and True after updating budget)

Wrapped by user_must_exist().

Parameters:

• user_name (str) – name of the user

• may_query (bool) – flag give or remove access to user

update_budget(user_name: str, dataset_name: str, spent_epsilon: float, spent_delta: float) → None

Update current epsilon and delta delta spent by user.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• spent_epsilon (float) – value of epsilon spent on last query

• spent_delta (float) – value of delta spent on last query

update_delta(user_name: str, dataset_name: str, spent_delta: float) → None

Update spent delta spent by user with spent delta of the user.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• spent_delta (float) – value of delta spent on last query

update_epsilon(user_name: str, dataset_name: str, spent_epsilon: float) → None

Update spent epsilon by user with total spent epsilon.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• spent_epsilon (float) – value of epsilon spent on last query

abstract update_epsilon_or_delta(user_name: str, dataset_name: str, parameter: BudgetDBKey, spent_value: float) → None

Update current budget spent by user with spent budget.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• parameter (str) – One of BudgetDBKey

• spent_value (float) – spending of epsilon or delta on last query

abstract wipe() → None

Wipe the entire Database.

class lomas_server.admin_database.BudgetDBKey(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Key for selecting budget values in admin db for given.

dataset and user.

DELTA_INIT = 'initial_delta'

DELTA_SPENT = 'total_spent_delta'

EPSILON_INIT = 'initial_epsilon'

EPSILON_SPENT = 'total_spent_epsilon'

class lomas_server.admin_database.LocalAdminDatabase(*, path: Path)

Bases: AdminDatabase

Local Admin database in a single file.

add_dataset(dataset_name: str, database_type: str, metadata_database_type: str, dataset_path: str | None = '', metadata_path: str = '', bucket: str | None = '', key: str | None = '', endpoint_url: str | None = '', credentials_name: str | None = '', metadata_bucket: str | None = '', metadata_key: str | None = '', metadata_endpoint_url: str | None = '', metadata_access_key_id: str | None = '', metadata_secret_access_key: str | None = '', metadata_credentials_name: str | None = '') → None

Set a database type to a dataset in dataset collection.

Parameters:

• dataset_name (str) – Dataset name

• database_type (str) – Type of the database

• metadata_database_type (str) – Metadata database type

• dataset_path (str) – Path to the dataset (for local db type)

• metadata_path (str) – Path to metadata (for local db type)

• bucket (str) – S3 bucket name

• key (str) – S3 key

• endpoint_url (str) – S3 endpoint URL

• credentials_name (str) – The name of the credentials in the server config to retrieve the dataset from S3 storage.

• metadata_bucket (str) – Metadata S3 bucket name

• metadata_key (str) – Metadata S3 key

• metadata_endpoint_url (str) – Metadata S3 endpoint URL

• metadata_access_key_id (str) – Metadata AWS access key ID

• metadata_secret_access_key (str) – Metadata AWS secret access key

• metadata_credentials_name (str) – The name of the credentials in the server config for retrieving the metadata.

Raises:

ValueError – If the dataset already exists or if the database type is unknown.

Returns:

None

add_dataset_to_user(username: str, dataset_name: str, epsilon: float, delta: float) → None

add_datasets_via_yaml(yaml_file: Path, clean: bool, path_prefix: str = '') → None

Set all database types to datasets in dataset collection based.

on yaml file.

Parameters:

• Path (yaml_file) – path to the YAML file location

• clean (bool) – Whether to clean the collection before adding.

• path_prefix (str, optional) – Prefix to add to all file paths. Defaults to “”.

Raises:

ValueError – If there are errors in the YAML file format.

Returns:

None

add_user(username: str, email: str, dataset_name: str | None = None, epsilon: float = 0.0, delta: float = 0.0) → None

Add new user in users collection with default values for all fields.

Parameters:

• username (str) – username to be added

• email (str) – email to be added

Raises:

• ValueError – If the username already exists.

• WriteConcernError – If the result is not acknowledged.

Returns:

None

add_users_via_yaml(yaml_file: Path, clean: bool) → None

Add all users from yaml file to the user collection.

Parameters:

• yaml_file (Path) – a path to the YAML file location

• clean (bool) – boolean flag True if drop current user collection False if keep current user collection

Returns:

None

datasets() → list[DSInfo]

del_dataset(dataset_name: str) → None

del_dataset_to_user(username: str, dataset_name: str) → None

del_user(username: str) → None

does_dataset_exist(dataset_name: str) → bool

Checks if dataset exist in the database.

Parameters:

dataset_name (str) – name of the dataset to check

Returns:

True if the dataset exists, False otherwise.

Return type:

bool

does_user_exist(user_name: str) → bool

Checks if user exist in the database.

Parameters:

user_name (str) – name of the user to check

Returns:

True if the user exists, False otherwise.

Return type:

bool

drop_archive() → None

drop_collection(collection: str) → None

get_and_set_may_user_query(user_name: str, may_query: bool) → bool

Atomic operation to check and set if the user may query the server.

(Set False before querying and True after updating budget)

Wrapped by user_must_exist().

Parameters:

• user_name (str) – name of the user

• may_query (bool) – flag give or remove access to user

Returns:

The may_query status of the user before the update.

Return type:

bool

get_archives_of_user(username: str) → list[dict]

get_collection(collection: str) → dict[str, Any]

get_dataset(dataset_name: str) → DSInfo

Get dataset access info based on dataset_name.

Wrapped by dataset_must_exist().

Parameters:

dataset_name (str) – Name of the dataset.

Returns:

The dataset model.

Return type:

Dataset

get_dataset_metadata(dataset_name: str) → Metadata

Returns the metadata dictionnary of the dataset.

Wrapped by dataset_must_exist().

Parameters:

dataset_name (str) – name of the dataset to get the metadata

Returns:

The metadata object.

Return type:

Metadata

get_epsilon_or_delta(user_name: str, dataset_name: str, parameter: BudgetDBKey) → float

Get the total spent epsilon or delta by user on dataset.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• parameter (str) – Member of BudgetDBKey.

Returns:

The requested budget value.

Return type:

float

get_user_previous_queries(user_name: str, dataset_name: str) → list[dict]

Retrieves and return the queries already done by a user.

Wrapped by user_must_have_access_to_dataset().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

List of previous queries.

Return type:

List[dict]

has_user_access_to_dataset(user_name: str, dataset_name: str) → bool

Checks if a user may access a particular dataset.

Wrapped by user_must_exist().

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

Returns:

True if the user has access, False otherwise.

Return type:

bool

load_dataset_collection(datasets: list[DSInfo], path_prefix: str) → None

load_users_collection(users: list[User]) → None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(_: Any, /) → None

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

path: Path

Database accepts existing path or new (creatable) path.

save_query(user_name: str, query: LomasRequestModel, response: QueryResponse) → None

Save queries of user on datasets in a separate collection (table).

Parameters:

• user_name (str) – name of the user

• query (LomasRequestModel) – Request object received from client

• response (QueryResponse) – Response object sent to client

update_epsilon_or_delta(user_name: str, dataset_name: str, parameter: BudgetDBKey, spent_value: float) → None

Update current budget spent by user with spent budget.

Parameters:

• user_name (str) – name of the user

• dataset_name (str) – name of the dataset

• parameter (str) – One of BudgetDBKey

• spent_value (float) – spending of epsilon or delta on last query

users() → list[User]

wipe() → None

Wipe the entire Database.