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.
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file arcane_flask-4.0.0.tar.gz.
File metadata
- Download URL: arcane_flask-4.0.0.tar.gz
- Upload date:
- Size: 14.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
poetry/2.4.1 CPython/3.12.13 Linux/6.17.0-1020-azure
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
985684d3077c6ed64176a2aeccf96480de2722babdea919f03e40488ccfa0188
|
|
| MD5 |
bac0d4036d1c38c8d0c89e632bcd9da1
|
|
| BLAKE2b-256 |
dd17dcb04d35cd9c6e7d473bd7820716c5c7140b7a4327fe080d632907572711
|
File details
Details for the file arcane_flask-4.0.0-py3-none-any.whl.
File metadata
- Download URL: arcane_flask-4.0.0-py3-none-any.whl
- Upload date:
- Size: 17.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
poetry/2.4.1 CPython/3.12.13 Linux/6.17.0-1020-azure
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
71628100276172036ef37c536477beab0441f4b075f5af587e21cca7fc40ddf4
|
|
| MD5 |
4a34f5ea12eefe00f1d175893849ede5
|
|
| BLAKE2b-256 |
b0d2801df503385440d813d1d4e4f9a2852641da08209ace976b44a2883bb363
|
