Skip to main content

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

Project description

Athena Client Library

PyPI - Version

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'

Project details


Release history Release notifications | RSS feed

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

resolver_athena_client-1.3.1.tar.gz (30.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

resolver_athena_client-1.3.1-py3-none-any.whl (41.9 kB view details)

Uploaded Python 3

File details

Details for the file resolver_athena_client-1.3.1.tar.gz.

File metadata

  • Download URL: resolver_athena_client-1.3.1.tar.gz
  • Upload date:
  • Size: 30.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: uv/0.10.9 {"installer":{"name":"uv","version":"0.10.9","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}

File hashes

Hashes for resolver_athena_client-1.3.1.tar.gz
Algorithm Hash digest
SHA256 abc54d4dae04e3ee2b0eb674c42124724123e166fbf0fafa9382adfcb6fc650a
MD5 abb98df8155c74c08a07a06b76dd4e94
BLAKE2b-256 288e4a9012a13cab06b65eabfbdaec3491a876c46f56088764341d76dfbb10dd

See more details on using hashes here.

File details

Details for the file resolver_athena_client-1.3.1-py3-none-any.whl.

File metadata

  • Download URL: resolver_athena_client-1.3.1-py3-none-any.whl
  • Upload date:
  • Size: 41.9 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: uv/0.10.9 {"installer":{"name":"uv","version":"0.10.9","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}

File hashes

Hashes for resolver_athena_client-1.3.1-py3-none-any.whl
Algorithm Hash digest
SHA256 6d789f605d1342cd80faf9cb21ddf272322565d56809aa4c9f1f57fffdd0c0f4
MD5 ce4b2e37ac1de57c9f7a6104ec66fca3
BLAKE2b-256 1e43bcdae01d687bcf80908894268982779104abe7009c8f5fd4f267544bf1d8

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page