arcane-flask 3.1.0

Utility functions for flask apps.

Arcane flask

This package help us authenticate users

Get Started

pip install arcane-flask

Example Usage

from arcane.flask import check_access_rights
from arcane.core import RightsLevelEnum, UserRightsEnum
from arcane.datastore import Client as DatastoreClient
from arcane.pubsub import Client as PubSubClient

datastore_client = DatastoreClient()
pubsub_client = PubSubClient()

@check_access_rights(
    service='my-service',
    required_rights=RightsLevelEnum.VIEWER,
    service_user_right=UserRightsEnum.MY_SERVICE,
    datastore_client=datastore_client,
    pubsub_client=pubsub_client,
    receive_rights_per_client=True,
    project='my-project',
    timeout=30  # Timeout in seconds
)
def function(params):
    pass

Timeout Configuration

The check_access_rights decorator includes a built-in timeout parameter that allows you to set a maximum execution time for the decorated function. If the function exceeds this timeout, the decorator will automatically return a timeout response.

import time

@check_access_rights(
    service='my-service',
    required_rights=RightsLevelEnum.VIEWER,
    service_user_right=UserRightsEnum.MY_SERVICE,
    datastore_client=datastore_client,
    pubsub_client=pubsub_client,
    project='my-project',
    timeout=2  # 2 seconds timeout
)
def slow_function():
    time.sleep(5)  # This will exceed the timeout
    return {'result': 'success'}

If the function exceeds the timeout, the decorator will return:

({'detail': 'Request Timeout'}, 504)

Note: Set the timeout parameter to a value a few seconds less than your Cloud Run or server timeout configuration to ensure proper error handling.

Activity tracking

Tracking records request usage (user, service, function, status code, execution time, optional metadata) and sends it to a configurable sink via a ports & adapters setup. You pass a tracker (adapter) and a service name at startup; every request is then tracked automatically.

Register tracking at app startup

Use init_tracking(app, tracker, service_name). The tracker is any implementation of TrackingPort (e.g. LogTrackingAdapter or PubSubTrackingAdapter).

Log adapter (development or debugging; events go to the Python logger at INFO):

from arcane.flask import init_tracking
from arcane.flask.tracking import LogTrackingAdapter

app = Flask(__name__)
tracker = LogTrackingAdapter()
init_tracking(app, tracker, service_name="my-service")

PubSub adapter (production; events go to a Google Cloud Pub/Sub topic):

from arcane.flask import init_tracking
from arcane.flask.tracking import PubSubTrackingAdapter
from arcane.pubsub import Client as PubSubClient

app = Flask(__name__)
pubsub_client = PubSubClient()
tracker = PubSubTrackingAdapter(
    pubsub_client=pubsub_client,
    project="my-project",
    topic="activity-tracker",  # optional; default is "activity-tracker"
)
init_tracking(app, tracker, service_name="my-service")

Custom columns (PubSub only) By default the PubSub adapter sends all standard fields. You can restrict which fields are published by passing columns:

tracker = PubSubTrackingAdapter(
    pubsub_client=pubsub_client,
    project="my-project",
    topic="activity-tracker",
    columns=["user_email", "service", "status_code", "execution_time"],
)

Undecorated routes are tracked

Routes that do not use check_access_rights are still tracked. The middleware fills in a baseline event using the request: function_name comes from Flask’s request.endpoint (the view function name), service from the service_name you passed to init_tracking, and email is empty.

Enriching context during the request

Use enrich_tracking() inside any route (with or without check_access_rights) to add or merge key/value pairs into the event’s additional_info. Multiple calls in the same request are merged.

from arcane.flask.tracking import enrich_tracking

@app.route("/process")
def process():
    enrich_tracking({"labels": ["feed"]})
    size = read_file_size()
    enrich_tracking({"file_size": size})
    return {"status": "ok"}, 200

The emitted event will include additional_info={"labels": ["feed"], "file_size": ...}.

Excluding endpoints

Pass excluded_endpoints so some views (e.g. health checks) are not tracked:

init_tracking(
    app,
    tracker,
    service_name="my-service",
    excluded_endpoints=frozenset({"health", "readiness"}),
)

Custom adapter

Implement the TrackingPort interface and pass it to init_tracking:

from arcane.flask.tracking import TrackingPort, ActivityEvent

class MyTrackingAdapter(TrackingPort):
    def emit(self, event: ActivityEvent) -> None:
        # send event to your backend (Kafka, HTTP, etc.)
        ...

Structured logging labels

The arcane.flask.logs module lets you add structured labels to every log entry in a request. This is useful for attaching metadata such as component, module, function, or IDs like optimization_id.

First, set up logging once at application startup:

from arcane.flask import logs

logs.setup_logging(gcp_project="my-gcp-project-id")

Then, register a teardown handler to clear labels at the end of each request:

from arcane.flask import logs

@app.teardown_request
def clear_logging_labels(exc):
    logs.clear_labels()

Inside your handlers you can either attach labels imperatively:

from arcane.flask import logs

def validate_model_parallelism_post(optimization_id: str, job_prefix: str, ...):
    logs.attach_labels(
        component="feed-boost",
        module="api",
        function="validate_model_parallelism_post",
        optimization_id=optimization_id,
        job_prefix=job_prefix,
    )
    ...

or use the with_log_labels decorator for static labels and still add dynamic labels as needed:

from arcane.flask import logs

@logs.with_log_labels(
    component="feed-boost",
    module="api",
    function="validate_model_parallelism_post",
)
def validate_model_parallelism_post(optimization_id: str, job_prefix: str, ...):
    logs.attach_labels(
        optimization_id=optimization_id,
        job_prefix=job_prefix,
    )
    ...

All log records emitted during the request will then include a logging.googleapis.com/labels field with the JSON-encoded labels.