# Python SDK

The Python SDK supports Python 3.10+ and has no runtime dependencies. It includes synchronous, thread-safe API calls, streamed uploads and downloads, safe retries, webhook verification, type hints for every method and an in-memory test client.

## Choose a moderation flow

Use the SDK's moderate method to wait for a decision in the request, or its create method to queue moderation and let your app continue. Text and images support both flows. Audio and video always run asynchronously, even when using moderate.

For queued content, save the returned asset ID and [set up a webhook](/docs/guides/webhooks) for [`asset.classified`](/docs/api/events#event-asset-classified) (decision ready) and [`asset.failed`](/docs/api/events#event-asset-failed) (processing failed). Keep content unpublished until approved. You can retrieve the asset periodically instead of receiving events. The [quick start](/docs) demonstrates both flows.

## Install

> The Python SDK is not yet available from PyPI.

## Moderate text

Keep credentials on your server. Without arguments, the client reads `CALMLENS_API_KEY` and `CALMLENS_PROJECT_ID` from the environment.

```python
from calmlens import CalmLensClient

client = CalmLensClient()
asset = client.text.moderate(text="Great photo!", external_id="comment-42")
if asset["status"] == "approved":
    print("Approved", (asset.get("report") or {}).get("flagged"))
```

Methods take snake\_case keyword arguments, or one options dict with the API's camelCase keys, such as `client.assets.retrieve({"assetId": asset_id})`. Responses are dicts with the API's camelCase field names.

Only publish content after an explicit `approved` result. An approved, flagged asset can still need review. Pending, rejected and failed requests must not be treated as approval. See [Report format](/docs/guides/moderation-reports) and [Policies and actions](/docs/guides/moderation-policies).

## Upload and deliver images

```python
with open("photo.jpg", "rb") as file:
    queued = client.assets.create(
        name="photo.jpg",
        content={"data": file, "type": "image/jpeg"},
        visibility="private",
        processing_options={"keep_content_after_processing": True},
    )

# Retrieve later, or respond to an asset.classified webhook.
asset = client.assets.retrieve(asset_id=queued["id"])
if asset["status"] == "approved":
    signed = client.assets.create_signed_url(asset_id=asset["id"], resource="content", expires_in_seconds=60)
    print(signed["url"])
```

`assets.create()` queues processing. `assets.moderate()` waits for a decision for supported content; audio and video remain asynchronous.

`data` accepts bytes or a binary file object. Files are streamed in chunks rather than read into memory, and their size is measured automatically. For a stream that can't seek, such as a pipe, pass its `sizeInBytes`. The timeout applies to a silent connection, not to the whole transfer, so large uploads and downloads are not cut off.

Remote downloads are limited to 10 MB for images and 20 MB for other files. For multipart limits and sequential uploads, see [Multipart uploads](/docs/guides/multipart-uploads). Retain content only when you need previews, transforms or CDN delivery.

## Operations

| Method | Purpose |
| --- | --- |
| `assets.create(...)` / `assets.moderate(...)` | Queue an upload or wait for classification. |
| `assets.retrieve(...)` / `assets.update(...)` | Read or update an asset. |
| `assets.list(...)` / `assets.iter(...)` | List one page of assets, or iterate over every page. |
| `assets.list_children(...)` / `assets.iter_children(...)` | List an asset's extracted children. |
| `assets.search(...)` | Search assets by keyword or meaning. |
| `assets.upload_part(...)` | Continue a sequential multipart upload. |
| `assets.move(...)` / `assets.retry(...)` | Move an asset or retry an eligible failure. |
| `assets.delete_content(...)` / `assets.delete(...)` | Remove stored content and keep the record, or delete the asset. |
| `assets.create_signed_url(...)` | Sign private CDN access. |
| `folders.list()` | List folders. |
| `folders.create/retrieve/update/move/delete(...)` | Manage folders. |
| `events.list(...)` / `events.iter(...)` | Read the project event feed. |
| `ensure_lens(...)` / `ensure_lenses(...)` | Ensure versioned lens definitions. |
| `cdn.get_asset_url(...)` / `cdn.serve(...)` | Build a CDN URL or stream content. |

List pagination starts at zero. Use `None` to clear nullable fields or move assets to the project root; omit a field to leave it unchanged.

## Concurrency and async code

One client is safe to share between threads, and reusing it keeps connections open between calls. In a forked worker process, such as a gunicorn or Celery worker, the client opens new connections rather than sharing the parent's. Calls are synchronous; from async code, run them in a thread:

```python
asset = await asyncio.to_thread(client.text.moderate, text=body, external_id=comment_id)
```

## Errors and retries

Every failure raises `CalmLensError` or one of its subclasses: `CalmLensAPIError` for API responses, `CalmLensTimeoutError` and `CalmLensTransportError` for network failures, and `CalmLensValidationError` for invalid arguments or configuration, detected before anything is sent. Each error has a stable `code` (constants in `calmlens.error_codes`), the HTTP `status`, the response's `request_id` to quote to support, `retry_after`, `headers`, and the `is_not_found` and `is_retryable` helpers.

```python
from calmlens import CalmLensAPIError, error_codes

try:
    asset = client.assets.retrieve(asset_id="asset-id")
except CalmLensAPIError as error:
    if error.code != error_codes.ASSET_NOT_FOUND:
        print(error.code, error.status, error.request_id)
        raise
    asset = None
```

The client retries reads, and uploads, after network failures, timeouts and <code>429</code>, <code>502</code>, <code>503</code> or <code>504</code> responses, up to `max_retries` times (default 2) with a growing random delay, or after the `Retry-After` delay. Uploads send an `Idempotency-Key`, generated per call and reused for its retries, so a retried upload never creates a second asset; pass `idempotency_key=` to retry from another process. Other writes are not retried. Configure `timeout` (default 60 seconds of inactivity), `connect_timeout` (default 10 seconds) and `max_retries` on the client, or per call with `client.with_options(timeout=300)`.

## Moderation results

`assets.moderate` and `text.moderate` return the asset as a `dict` with the API's camelCase keys. `id` and `status` are always present. `report` is missing or `None` until classification finishes, so read it with `asset.get("report")`. Other optional fields, such as `externalId`, `folderId` and `tags`, appear only when they apply. Status and other enum values are plain strings, so a value added in a newer API version doesn't break your code; treat any status other than `approved` as not approved. See [Report format](/docs/guides/moderation-reports).

## Events feed

`events.list` returns `{"events": [...], "cursor": ..., "hasMore": ...}`. Without `after`, it returns no events and a cursor for the current time. `events.iter_pages` follows the cursor from `after` until `hasMore` is `False`; save the last page's cursor to resume.

```python
cursor = client.events.list()["cursor"]  # no events; starts from now
for page in client.events.iter_pages(after=cursor, limit=50):
    for event in page["events"]:
        print(event["type"], event["facts"])
    cursor = page["cursor"]
```

## Webhooks

Verify the exact raw request body, then parse it, with `parse_webhook_event`. It accepts only the raw `str` or `bytes` body: a body that was parsed and serialised again rarely has the same bytes, so the signature would not match.

```python
import os
from calmlens import WebhookVerificationError, parse_webhook_event

def handle_webhook(raw_body: bytes, headers) -> int:
    try:
        event = parse_webhook_event(raw_body, headers=headers, secret=os.environ["CALMLENS_WEBHOOK_SECRET"])
    except WebhookVerificationError:
        return 401
    # Queue the work, keyed by event["id"], then respond quickly.
    return 204
```

Pass the raw body your framework received:

| Framework | Raw body | Headers |
| --- | --- | --- |
| Flask | `request.get_data()` | `request.headers` |
| Django | `request.body` (exempt the view from CSRF) | `request.headers` |
| FastAPI or Starlette | `await request.body()` (don't declare a body model) | `request.headers` |

`verify_webhook_signature` takes the same arguments and returns `True` or `False`. Both read the `x-calmlens-signature` and `x-calmlens-timestamp` headers from `headers`, or take them as `signature=` and `timestamp=`. Timestamps more than 5 minutes old are rejected. Deliveries also carry `x-calmlens-delivery`, a unique ID for deduplicating retries. See [Webhooks & Connectors](/docs/guides/webhooks).

Install the standalone [CalmLens CLI](/docs/guides/local-development) to forward events locally:

```bash
calmlens listen --forward-to http://localhost:3000/webhook --env-file .env.local
calmlens trigger clean
```

## Test without API calls

See [Python testing](/docs/sdk/python/testing) for offline mock setup, request inspection and simulated failures.