lomas_server.admin_database package

Submodules

lomas_server.admin_database.admin_database module

class lomas_server.admin_database.admin_database.AdminDatabase(**connection_parameters: Dict[str, str])[source]

Bases: ABC

Overall database management for server state.

This is an abstract class.

abstract does_dataset_exist(dataset_name: str) → bool[source]

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[source]

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[source]

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_field(dataset_name: str, key: str) → str[source]

Get dataset field type based on dataset name and key

Wrapped by dataset_must_exist().

Parameters:

  • dataset_name (str) – Name of the dataset.

  • key (str) – Key for the value to get in the dataset dict.

Returns:

The requested value.

Return type:

str

abstract get_dataset_metadata(dataset_name: str) → Metadata[source]

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: str) → float[source]

Get the total spent epsilon or delta by a specific user on a specific dataset

Parameters:

  • user_name (str) – name of the user

  • dataset_name (str) – name of the dataset

  • parameter (str) – total_spent_epsilon or total_spent_delta

Returns:

The requested budget value.

Return type:

float

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

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][source]

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][source]

Get the total spent epsilon and delta spent by a specific user on a specific dataset (since the initialisation)

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][source]

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[source]

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

prepare_save_query(user_name: str, query_json: RequestModel, response: dict) → dict[source]

Prepare the query to save in archives

Parameters:

  • user_name (str) – name of the user

  • query_json (RequestModel) – request received from client

  • response (dict) – response sent to the 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_json: RequestModel, response: dict) → None[source]

Save queries of user on datasets in a separate collection (table) named “queries_archives” in the DB

Parameters:

  • user_name (str) – name of the user

  • query_json (dict) – json received from client

  • response (dict) – response sent to the client

abstract set_may_user_query(user_name: str, may_query: bool) → bool[source]

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[source]

Update the current epsilon and delta spent by a specific user with the last spent delta

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[source]

Update the spent delta spent by a specific user with the total 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[source]

Update the spent epsilon by a specific user with the 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: str, spent_value: float) → None[source]

Update the current budget spent by a specific user with the last spent budget.

Parameters:

  • user_name (str) – name of the user

  • dataset_name (str) – name of the dataset

  • parameter (str) – “current_epsilon” or “current_delta”

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

lomas_server.admin_database.admin_database.dataset_must_exist(func: Callable) → Callable[source]

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) → Callable[source]

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) → Callable[source]

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.factory module

lomas_server.admin_database.factory.admin_database_factory(config: DBConfig) → AdminDatabase[source]

Instantiates and returns the correct database type described in the provided config.

Parameters:

config (DBConfig) – An instance of DBconfig.

Raises:
Returns:

A instance of the correct type of AdminDatabase.

Return type:

AdminDatabase

lomas_server.admin_database.mongodb_database module

class lomas_server.admin_database.mongodb_database.AdminMongoDatabase(connection_string: str, database_name: str)[source]

Bases: AdminDatabase

Overall MongoDB database management for server state.

does_dataset_exist(dataset_name: str) → bool[source]

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[source]

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

get_and_set_may_user_query(user_name: str, may_query: bool) → bool[source]

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_dataset_field(dataset_name: str, key: str) → str[source]

Get dataset field type based on dataset name and key

Wrapped by dataset_must_exist().

Parameters:

  • dataset_name (str) – Name of the dataset.

  • key (str) – Key for the value to get in the dataset dict.

Returns:

The requested value.

Return type:

str

get_dataset_metadata(dataset_name: str) → Metadata[source]

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 model.

Return type:

Metadata

get_epsilon_or_delta(user_name: str, dataset_name: str, parameter: str) → float[source]

Get the total spent epsilon or delta by a specific user on a specific dataset

Parameters:

  • user_name (str) – name of the user

  • dataset_name (str) – name of the dataset

  • parameter (str) – total_spent_epsilon or total_spent_delta

Returns:

The requested budget value.

Return type:

float

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

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[source]

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

save_query(user_name: str, query_json: RequestModel, response: dict) → None[source]

Save queries of user on datasets in a separate collection (table) named “queries_archives” in the DB

Parameters:

  • user_name (str) – name of the user

  • query_json (RequestModel) – json received from client

  • response (dict) – response sent to the client

Raises:

WriteConcernError – If the result is not acknowledged.

set_may_user_query(user_name: str, may_query: bool) → None[source]

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

Raises:

WriteConcernError – If the result is not acknowledged.

update_epsilon_or_delta(user_name: str, dataset_name: str, parameter: str, spent_value: float) → None[source]

Update the current budget spent by a specific user with the last spent budget.

Parameters:

  • user_name (str) – name of the user

  • dataset_name (str) – name of the dataset

  • parameter (str) – “current_epsilon” or “current_delta”

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

Raises:

WriteConcernError – If the result is not acknowledged.

lomas_server.admin_database.mongodb_database.check_result_acknowledged(res: _WriteResult) → None[source]

Raises an exception if the result is not acknowledged.

Parameters:

res (_WriteResult) – The PyMongo WriteResult to check.

Raises:

WriteConcernError – If the result is not acknowledged.

lomas_server.admin_database.utils module

lomas_server.admin_database.utils.add_demo_data_to_mongodb_admin(user_yaml: str = '/data/collections/user_collection.yaml', dataset_yaml: str = '/data/collections/dataset_collection.yaml') → None[source]

Adds the demo data to the mongodb admindb. Meant to be used in the develop mode of the service.

Parameters:

  • user_yaml (str) – path to user collection yaml file

  • dataset_yaml (str) – path to dataset collection yaml file

lomas_server.admin_database.utils.get_mongodb() → Database[source]

Get URL of the administration MongoDB.

Parameters:

config (DBConfig) – An instance of DBConfig.

Returns:

A correctly formatted url for connecting to the

MongoDB database.

Return type:

str

lomas_server.admin_database.utils.get_mongodb_url(config: MongoDBConfig) → str[source]

Get URL of the administration MongoDB.

Parameters:

config (MongoDBConfig) – An instance of DBConfig.

Returns:

A correctly formatted url for connecting to the

MongoDB database.

Return type:

str

lomas_server.admin_database.yaml_database module

class lomas_server.admin_database.yaml_database.AdminYamlDatabase(yaml_db_path: str)[source]

Bases: AdminDatabase

Overall Yaml database management for server state

does_dataset_exist(dataset_name: str) → bool[source]

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[source]

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

get_and_set_may_user_query(user_name: str, may_query: bool) → bool[source]

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_dataset_field(dataset_name: str, key: str) → str[source]

Get dataset field type based on dataset name and key

Wrapped by dataset_must_exist().

Parameters:

  • dataset_name (str) – Name of the dataset.

  • key (str) – Key for the value to get in the dataset dict.

Returns:

The requested value.

Return type:

str

get_dataset_metadata(dataset_name: str) → Metadata[source]

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 model.

Return type:

Metadata

get_epsilon_or_delta(user_name: str, dataset_name: str, parameter: str) → float[source]

Get the total spent epsilon or delta by a specific user on a specific dataset

Parameters:

  • user_name (str) – name of the user

  • dataset_name (str) – name of the dataset

  • parameter (str) – total_spent_epsilon or total_spent_delta

Returns:

The requested budget value.

Return type:

float

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

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[source]

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

save_current_database() → None[source]

Saves the current database with updated parameters in new yaml with the date and hour in the path Might be useful to verify state of DB during development

save_query(user_name: str, query_json: RequestModel, response: dict) → None[source]

Save queries of user on datasets in a separate collection (table) named “queries_archives” in the DB

Parameters:

  • user_name (str) – name of the user

  • query_json (RequestModel) – request received from client

  • response (dict) – response sent to the client

set_may_user_query(user_name: str, may_query: bool) → None[source]

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_epsilon_or_delta(user_name: str, dataset_name: str, parameter: str, spent_value: float) → None[source]

Update the current budget spent by a specific user with the last spent budget.

Parameters:

  • user_name (str) – name of the user

  • dataset_name (str) – name of the dataset

  • parameter (str) – “current_epsilon” or “current_delta”

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

Module contents