resolver_athena_client 1.2.1.dev20260306165521

Python client library for Athena API - CSAM detection and content classification

Athena Client Library

This is a Python library for interacting with the Athena API (Resolver Unknown CSAM Detection).

Authentication

The Athena client supports two authentication methods:

Static Token Authentication

from resolver_athena_client.client.channel import create_channel

# Use a pre-existing authentication token
channel = create_channel(host="your-host", auth_token="your-token")

OAuth Credential Helper (Recommended)

The credential helper automatically handles OAuth token acquisition and refresh:

import asyncio
from resolver_athena_client.client.channel import CredentialHelper, create_channel_with_credentials

async def main():
    # Create credential helper with OAuth settings
    credential_helper = CredentialHelper(
        client_id="your-oauth-client-id",
        client_secret="your-oauth-client-secret",
        auth_url="https://crispthinking.auth0.com/oauth/token",  # Optional, this is default
        audience="crisp-athena-live"  # Optional, this is default
    )

    # Create channel with automatic OAuth handling
    channel = await create_channel_with_credentials(
        host="your-host",
        credential_helper=credential_helper
    )

asyncio.run(main())

Environment Variables

For the OAuth example to work, set these environment variables:

export OAUTH_CLIENT_ID="your-client-id"
export OAUTH_CLIENT_SECRET="your-client-secret"
export ATHENA_HOST="your-athena-host"

OAuth Features

• Automatic token refresh: Tokens are automatically refreshed when they expire
• Thread-safe: Multiple concurrent requests will safely share cached tokens
• Error handling: Comprehensive error handling for OAuth failures
• Configurable: Custom OAuth endpoints and audiences supported

See examples/oauth_example.py for a complete working example.

Examples

• examples/example.py - Basic classification example with static token
• examples/oauth_example.py - OAuth authentication with credential helper
• examples/create_image.py - Image generation utilities

TODO

Async pipelines

Make pipeline style invocation of the async interators such that we can

async read file -> async transform -> async classify -> async results

More async pipeline transformers

Add additional pipeline transformers for:

• Image format conversion
• Metadata extraction
• Error recovery and retry

Development

This package uses uv to manage its packages.

To install dependencies, run:

uv sync --dev

To build the package, run:

uv build

To run the standard tests, run:

pytest -m 'not functional'

Developers wishing to run the functional tests should see the Functional Tests section below.

To lint and format the code, run:

ruff check
ruff format

There are pre-commit hooks that will lint, format, and type check the code. Install them with:

pre-commit install

To re-compile the protobuf files, run from the repository's root directory:

bash scripts/compile_proto.sh

Functional Tests

Functional tests require an Athena environment to run against.

Pre-Requisites

You will need:

• An Athena host URL.
• An OAuth client ID and secret with access to the Athena environment.
• An affiliate with Athena enabled.

Preparing your environment

You can set up the environment variables in a .env file in the root of the repository, or in your shell environment:

You must set the following variables:

ATHENA_HOST=your-athena-host (e.g. localhost:5001)
ATHENA_TEST_AFFILIATE=your-affiliate-id
OAUTH_CLIENT_ID=your-oauth-client-id
OAUTH_CLIENT_SECRET=your-oauth-client-secret
ATHENA_TEST_PLATFORM_TOKEN=a standard platform token - this should be rejected
as only athena specific tokens are accepted.
ATHENA_TEST_EXPIRED_TOKEN=a valid but expired token - this should be rejected.

You can optionally set the following variables:

OAUTH_AUTH_URL=your-oauth-auth-url (default: https://crispthinking.auth0.com/oauth/token)
OAUTH_AUDIENCE=your-oauth-audience (default: crisp-athena-live)
TEST_IMAGE_COUNT=number-of-images-to-test-with (default: 5000) - this is the
number of images the _streaming_ test will use.
TEST_MIN_INTERVAL=minimum-interval-in-ms (default: None, send as fast as
possible) - this is the minimum interval between
images for the _streaming_ test.
ATHENA_NON_EXISTENT_AFFILIATE=non-existent-affiliate-id (default:
thisaffiliatedoesnotexist123) - this is used to test error handling.
ATHENA_NON_PERMITTED_AFFILIATE=non-permitted-affiliate-id (default:
thisaffiliatedoesnothaveathenaenabled) - this is used to test error handling.
ATHENA_E2E_TESTCASE_DIR=test-case-directory (default: integrator_sample) - this is the test case directory to use for the e2e tests.
See E2E Tests section below for more details.

Then run the functional tests with:

pytest -m functional

E2E Tests

The e2e tests assert that the API returns some expected scores rather than exercising different API paths. As such, they are dependent on the classifier that you are calling through the API. Right now, there are 2 types of classifier, benign and live. By default, the tests will run the integrator_sample test set, which uses the live classifier. If you wish to use the benign classifier instead, you may set the ATHENA_E2E_TESTCASE_DIR environment variable to benign_model.

Alternatively, you may disable these tests altogether, by excluding tests that have the e2e marker, something like this:

pytest -m 'functional and not e2e'