Metadata-Version: 2.4
Name: arcane-flask
Version: 3.1.0
Summary: Utility functions for flask apps.
Author: Arcane
Author-email: product@wearcane.com
Requires-Python: >=3.8,<4.0
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Dist: arcane-core (>=1.6.0,<2.0.0)
Requires-Dist: arcane-datastore (>=1.1.0,<2.0.0)
Requires-Dist: arcane-pubsub (>=1.1.0,<2.0.0)
Requires-Dist: backoff (>=1.10.0)
Requires-Dist: firebase_admin (==5.2.0)
Requires-Dist: flask (>=2.2.0,<3.0.0)
Requires-Dist: flask_log_request_id (>=0.10.1,<0.11.0)
Requires-Dist: pydash (>=8.0.3,<9.0.0)
Requires-Dist: pytz (>=2024.2,<2025.0)
Description-Content-Type: text/markdown

# Arcane flask

This package help us authenticate users

## Get Started

```sh
pip install arcane-flask
```

## Example Usage

```python
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.

```python
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:

```python
({'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):

```python
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):

```python
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`:

```python
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.

```python
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:

```python
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`:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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.

