Metadata-Version: 2.3
Name: resolver_athena_client
Version: 1.3.1.dev20260306171415
Summary: Python client library for Athena API - CSAM detection and content classification
Keywords: csam,content-detection,image-classification,api-client,grpc
Author: Crisp Thinking Group Ltd
Author-email: Crisp Thinking Group Ltd <opensource@kroll.com>
License: MIT
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Dist: anyio>=4.10.0
Requires-Dist: brotli>=1.1.0
Requires-Dist: grpcio>=1.78.0,!=1.78.1,<2.0.0
Requires-Dist: httpx>=0.25.0
Requires-Dist: numpy>=2.2.6
Requires-Dist: opencv-python-headless>=4.13.0.92
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# Athena Client Library

![PyPI - Version](https://img.shields.io/pypi/v/resolver-athena-client)

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

```python
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:
```bash
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](https://docs.astral.sh/uv/) to manage its packages.

To install dependencies, run:

```bash
uv sync --dev
```

To build the package, run:

```bash
uv build
```

To run the standard tests, run:

```bash
pytest -m 'not functional'
```

Developers wishing to run the functional tests should see the
[Functional Tests](#functional-tests) section below.


To lint and format the code, run:

```bash
ruff check
ruff format
```

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

```bash
pre-commit install
```

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

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

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

```bash
pytest -m 'functional and not e2e'
```
