lomas_server.routes package

Submodules

lomas_server.routes.middlewares module

class lomas_server.routes.middlewares.FastAPIMetricMiddleware(app: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]], app_name: str)[source]

Bases: BaseHTTPMiddleware

Middleware to collect and expose Prometheus metrics for a FastAPI application.

This middleware tracks various metrics related to HTTP requests, including: - Total requests (fastapi_requests_total) - Total responses (fastapi_responses_total) - Exceptions raised (fastapi_exceptions_total) - Request processing duration (fastapi_requests_duration_seconds) - Current requests in progress (fastapi_requests_in_progress)

It also supports integration with an OpenTelemetry exporter for exporting metrics to a metrics collector (e.g., Prometheus or any other OTLP-compatible collector).

async dispatch(request: Request, call_next: Callable[[Request], Awaitable[Response]]) → Response[source]

Processes HTTP request, records metrics and returns the HTTP response.

This method performs the following steps: 1. Tracks the current request in progress using fastapi_requests_in_progress gauge. 2. Records the request count with fastapi_requests_total counter. 3. Records the time taken to process the request using fastapi_requests_duration_seconds histogram. 4. Handles exceptions, if raised, and records the exception details using fastapi_exceptions_total counter. 5. Records the response status code with fastapi_responses_total counter. 6. Decrements the in-progress request gauge after processing.

Parameters:

  • request (Request) – The incoming HTTP request to be processed.

  • call_next (RequestResponseEndpoint) – Endpoint that processes the request and returns a response.

Returns:

The HTTP response after processing the request.

Return type:

Response

Raises:

BaseException – If an exception occurs during request processing, it is raised after logging it.

static get_path(request: Request) → tuple[str, bool][source]

Attempts to match the request’ route to a defined route.

Parameters:

request (Request) – The HTTP request to check for a matching path.

Returns:

A tuple containing:

  • The matched path (str) from the request URL.

  • Boolean (True if the path was handled by one of the routes).

Return type:

Tuple[str, bool]

class lomas_server.routes.middlewares.LoggingAndTracingMiddleware(app: Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]], dispatch: Callable[[Request, Callable[[Request], Awaitable[Response]]], Awaitable[Response]] | None = None)[source]

Bases: BaseHTTPMiddleware

Middleware for logging and tracing incoming HTTP requests.

This middleware logs the incoming requests, including the user name the route being accessed, and any query parameters. Additionally, it creates a trace span to trace the user’s request and adds attributes to the span related to the user name and query parameters.

async dispatch(request: Request, call_next: Callable[[Request], Awaitable[Response]]) → Response[source]

Handles the request and performs logging and tracing.

Logs the user name, the route and the query parameters. Creates a trace span to monitor the request and adds relevant attributes.

Parameters:

  • request (Request) – The incoming request object.

  • call_next (Callable) – A function that, when called, passes the request to the next middleware or request handler.

Returns:

The HTTP response generated by calling call_next(request).

Return type:

Response

lomas_server.routes.routes_admin module

lomas_server.routes.routes_admin.get_dataset_metadata(request: Request, user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], query_json: LomasRequestModel = Body({'dataset_name': 'PENGUIN'})) → Metadata[source]

Retrieves metadata for a given dataset.

Parameters:

  • request (Request) – Raw request object

  • user_id (UserId) – A UserId object identifying the user.

  • query_json (LomasRequestModel, optional) – A JSON object containing the dataset_name key for indicating the dataset. Defaults to Body(example_get_admin_db_data).

Raises:
Returns:

The metadata object for the specified

dataset_name.

Return type:

Metadata

lomas_server.routes.routes_admin.get_dummy_dataset(request: Request, user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], query_json: GetDummyDataset = Body({'dataset_name': 'PENGUIN', 'dummy_nb_rows': 100, 'dummy_seed': 42})) → DummyDsResponse[source]

Generates and returns a dummy dataset.

Parameters:

  • request (Request) – Raw request object

  • user_id (UserId) – A UserId object identifying the user.

  • query_json (GetDummyDataset, optional) –

    A JSON object containing the following:

    • nb_rows (int, optional): The number of rows in the dummy dataset (default: 100).

    • seed (int, optional): The random seed for generating the dummy dataset (default: 42).

    Defaults to Body(example_get_dummy_dataset).

Raises:
Returns:

a dict with the dataframe as a dict, the column types

and the list of datetime columns.

Return type:

JSONResponse

lomas_server.routes.routes_admin.get_initial_budget(request: Request, user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], query_json: LomasRequestModel = Body({'dataset_name': 'PENGUIN'})) → InitialBudgetResponse[source]

Returns the initial budget for a user and dataset.

Parameters:

  • request (Request) – Raw request object.

  • user_id (UserId) – A UserId object identifying the user.

  • query_json (LomasRequestModel, optional) –

    A JSON object containing: - dataset_name (str): The name of the dataset.

    Defaults to Body(example_get_admin_db_data).

Raises:
Returns:

a JSON object with:

  • initial_epsilon (float): initial epsilon budget.

  • initial_delta (float): initial delta budget.

Return type:

JSONResponse

lomas_server.routes.routes_admin.get_remaining_budget(request: Request, user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], query_json: LomasRequestModel = Body({'dataset_name': 'PENGUIN'})) → RemainingBudgetResponse[source]

Returns the remaining budget for a user and dataset.

Parameters:

  • request (Request) – Raw request object.

  • user_id (UserId) – A UserId object identifying the user.

  • query_json (LomasRequestModel, optional) –

    A JSON object containing: - dataset_name (str): The name of the dataset.

    Defaults to Body(example_get_admin_db_data).

Raises:
Returns:

a JSON object with:

  • remaining_epsilon (float): remaining epsilon budget.

  • remaining_delta (float): remaining delta budget.

Return type:

JSONResponse

async lomas_server.routes.routes_admin.get_server_config(_: Annotated[UserId, Security(get_user_id_from_authenticator)]) → ConfigResponse[source]

Returns the config of this server instance.

Parameters:

_ (UserId) – A UserId object identifying the user.

Returns:

The server config.

Return type:

ConfigResponse

async lomas_server.routes.routes_admin.get_state(_: Annotated[UserId, Security(get_user_id_from_authenticator)]) → JSONResponse[source]

Returns the current state dict of this server instance.

Parameters:

_ (UserId) – A UserId object identifying the user.

Returns:

The state of the server instance.

Return type:

JSONResponse

lomas_server.routes.routes_admin.get_total_spent_budget(request: Request, user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], query_json: LomasRequestModel = Body({'dataset_name': 'PENGUIN'})) → SpentBudgetResponse[source]

Returns the spent budget for a user and dataset.

Parameters:

  • request (Request) – Raw request object.

  • user_id (UserId) – A UserId object identifying the user.

  • query_json (LomasRequestModel, optional) –

    A JSON object containing: - dataset_name (str): The name of the dataset.

    Defaults to Body(example_get_admin_db_data).

Raises:
Returns:

a JSON object with:

  • total_spent_epsilon (float): total spent epsilon budget.

  • total_spent_delta (float): total spent delta budget.

Return type:

JSONResponse

lomas_server.routes.routes_admin.get_user_previous_queries(request: Request, user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], query_json: LomasRequestModel = Body({'dataset_name': 'PENGUIN'})) → JSONResponse[source]

Returns the query history of a user on a specific dataset.

Parameters:

  • request (Request) – Raw request object.

  • user_id (UserId) – A UserId object identifying the user.

  • query_json (LomasRequestModel, optional) –

    A JSON object containing: - dataset_name (str): The name of the dataset.

    Defaults to Body(example_get_admin_db_data).

Raises:
Returns:

A JSON object containing:

  • previous_queries (list[dict]): a list of dictionaries containing the previous queries.

Return type:

JSONResponse

async lomas_server.routes.routes_admin.health_handler() → JSONResponse[source]

HealthCheck endpoint: server alive.

Returns:

“live”

Return type:

JSONResponse

async lomas_server.routes.routes_admin.root() → RedirectResponse[source]

Redirect root endpoint to the state endpoint.

Returns:

The state of the server instance.

Return type:

JSONResponse

async lomas_server.routes.routes_admin.status_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, uid: UUID, response: Response) → Job[source]

Job status endpoint.

Parameters:

  • user_id (UserId) – The user id.

  • request (Request) – The raw request.

  • uid (UUID) – The job’s unique id.

  • response (Response) – The job status response.

Raises:
Returns:

The Job model for this uid.

Return type:

Job

lomas_server.routes.routes_dp module

async lomas_server.routes.routes_dp.diffprivlib_query_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, diffprivlib_query: DiffPrivLibQueryModel) → Job[source]

Handles queries for the DiffPrivLib Library.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • diffprivlib_query (DiffPrivLibQueryModel) – The diffprivlib query body.

Raises:
Returns:

a scheduled Job resulting in a QueryResponse containing a DiffPrivLibQueryResult.

Return type:

Job

async lomas_server.routes.routes_dp.dummy_diffprivlib_query_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, query_json: DiffPrivLibDummyQueryModel) → Job[source]

Handles queries on dummy datasets for the DiffPrivLib library.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • diffprivlib_query (DiffPrivLibDummyQueryModel) – The diffprivlib query body.

Raises:
Returns:

a scheduled Job resulting in a QueryResponse containing a DiffPrivLibQueryResult.

Return type:

Job

async lomas_server.routes.routes_dp.dummy_opendp_query_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, opendp_query: OpenDPDummyQueryModel) → Job[source]

Handles queries on dummy datasets for the OpenDP library.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object.

  • opendp_query (OpenDPQueryModel) – The opendp query object.

Raises:
Returns:

a scheduled Job resulting in a QueryResponse containing an OpenDPQueryResult.

Return type:

Job

async lomas_server.routes.routes_dp.dummy_smartnoise_sql_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, smartnoise_sql_query: SmartnoiseSQLDummyQueryModel) → Job[source]

Handles queries on dummy datasets for the SmartNoiseSQL library.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • smartnoise_sql_query (SmartnoiseSQLDummyQueryModel) – The smartnoise_sql query body.

Raises:
Returns:

a scheduled Job resulting in a QueryResponse containing a SmartnoiseSQLQueryResult.

Return type:

Job

async lomas_server.routes.routes_dp.dummy_smartnoise_synth_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, smartnoise_synth_query: SmartnoiseSynthDummyQueryModel) → Job[source]

Handles queries on dummy datasets for the SmartNoiseSynth library.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • smartnoise_synth_query (SmartnoiseSynthDummyQueryModel) – The smartnoise_synth query body.

Raises:
Returns:

a scheduled Job resulting in a QueryResponse containing a SmartnoiseSynthModel or SmartnoiseSynthSamples.

Return type:

Job

async lomas_server.routes.routes_dp.estimate_diffprivlib_cost(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, diffprivlib_query: DiffPrivLibRequestModel) → Job[source]

Estimates the privacy loss budget cost of an DiffPrivLib query.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • diffprivlib_query (DiffPrivLibRequestModel) – The diffprivlib query body.

  • following (A JSON object containing the) –

    • pipeline: The DiffPrivLib pipeline for the query.

    • feature_columns: the list of feature column to train

    • target_columns: the list of target column to predict

    • test_size: proportion of the test set

    • test_train_split_seed: seed for the random train test split,

    • imputer_strategy: imputation strategy

    Defaults to Body(example_dummy_diffprivlib).

Raises:
Returns:

a scheduled Job resulting in a CostResponse containing the privacy loss cost of the input query.

Return type:

Job

async lomas_server.routes.routes_dp.estimate_opendp_cost(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, opendp_query: OpenDPRequestModel) → Job[source]

Estimates the privacy loss budget cost of an OpenDP query.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object.

  • opendp_query (OpenDPRequestModel) – The opendp query object.

Raises:
Returns:

a scheduled Job resulting in a CostResponse containing the privacy loss cost of the input query.

Return type:

Job

async lomas_server.routes.routes_dp.estimate_smartnoise_sql_cost(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, smartnoise_sql_query: SmartnoiseSQLRequestModel) → Job[source]

Estimates the privacy loss budget cost of a SmartNoiseSQL query.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • smartnoise_sql_query (SmartnoiseSQLRequestModel) – The smartnoise_sql request body.

Raises:
Returns:

a scheduled Job resulting in a CostResponse containing the privacy loss cost of the input query.

Return type:

Job

async lomas_server.routes.routes_dp.estimate_smartnoise_synth_cost(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, smartnoise_synth_query: SmartnoiseSynthRequestModel) → Job[source]

Computes the privacy loss budget cost of a SmartNoiseSynth query.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • smartnoise_synth_query (SmartnoiseSynthRequestModel) – The smartnoise_synth query body.

Raises:
Returns:

a scheduled Job resulting in a CostResponse containing the privacy loss cost of the input query.

Return type:

Job

async lomas_server.routes.routes_dp.opendp_query_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, opendp_query: OpenDPQueryModel) → Job[source]

Handles queries for the OpenDP Library.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object.

  • opendp_query (OpenDPQueryModel) – The opendp query object.

Raises:
Returns:

a scheduled Job resulting in a QueryResponse containing an OpenDPQueryResult.

Return type:

Job

async lomas_server.routes.routes_dp.smartnoise_sql_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, smartnoise_sql_query: SmartnoiseSQLQueryModel) → Job[source]

Handles queries for the SmartNoiseSQL library.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • smartnoise_sql_query (SmartnoiseSQLQueryModel) – The smartnoise_sql query body.

Raises:
Returns:

a scheduled Job resulting in a QueryResponse containing a SmartnoiseSQLQueryResult.

Return type:

Job

async lomas_server.routes.routes_dp.smartnoise_synth_handler(user_id: Annotated[UserId, Security(get_user_id_from_authenticator)], request: Request, smartnoise_synth_query: SmartnoiseSynthQueryModel) → Job[source]

Handles queries for the SmartNoiseSynth library.

Parameters:

  • user_id (UserId) – A UserId object identifying the user.

  • request (Request) – Raw request object

  • smartnoise_synth_query (SmartnoiseSynthQueryModel) – The smartnoise_synth query body.

Raises:
Returns:

a scheduled Job resulting in a QueryResponse containing a SmartnoiseSynthModel or SmartnoiseSynthSamples.

Return type:

Job

lomas_server.routes.utils module

lomas_server.routes.utils.get_user_id_from_authenticator(request: Request, security_scopes: SecurityScopes, auth_creds: Annotated[HTTPAuthorizationCredentials, Depends(HTTPBearer)]) → UserId[source]

Extracts the authenticator from the app state and calls its get_user_id method.

Also adds the user_name to the request state to annotate the telemetry request span.

Parameters:

  • request (Request) – The request to access the app and state.

  • security_scopes (SecurityScopes) – The required scopes for the endpoint.

  • auth_creds (Annotated[HTTPAuthorizationCredentials, Depends) – The HTTP bearer token.

Returns:

A UserId instance extracted from the token.

Return type:

UserId

lomas_server.routes.utils.handle_query_to_job(request: Request, query: DummyQueryModel | QueryModel | LomasRequestModel, user_name: str, dp_library: DPLibraries) → Job[source]

Submit Job to handles queries on private, dummy and cost datasets on a worker.

Parameters:
Raises:

UnauthorizedAccessException – A query is already ongoing for this user, the user does not exist or does not have access to the dataset.

Returns:

A scheduled Job resulting in a QueryResponse containing the result of the query

(specific to the library) as well as the cost of the query. or a CostResponse containing the epsilon, delta and privacy-loss budget cost for the request.

Return type:

Job

async lomas_server.routes.utils.process_response(queue: Queue, cls: type[QueryResponse | CostResponse], jobs: dict[UUID, Job]) → None[source]

Process responses queue into Jobs.

async lomas_server.routes.utils.rabbitmq_connect_queue(config: Config, reconnect_interval: int = 10, timeout: int = 120) → RobustConnection[source]

Attempt with retries to connect to the queue.

lomas_server.routes.utils.rabbitmq_ctx(app: FastAPI) → AsyncIterator[None][source]

RabbitMQ queue context to connect and register callbacks.

lomas_server.routes.utils.timing_protection(func)[source]

Adds delays to requests response to protect against timing attack.

Module contents