Metadata-Version: 2.4
Name: authpi-admin
Version: 0.3.0
Summary: Official Python Admin SDK for AuthPI Core API
Project-URL: Homepage, https://authpi.com
Project-URL: Documentation, https://docs.authpi.com/sdk/python/admin
Project-URL: Repository, https://github.com/arbfay/authpi
Author-email: AuthPI <hello@authpi.com>
License: MIT
Keywords: admin,api,authentication,authorization,authpi,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic>=2.0.0
Provides-Extra: dev
Requires-Dist: mypy>=1.13.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24.0; extra == 'dev'
Requires-Dist: pytest-httpx>=0.34.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.8.0; extra == 'dev'
Description-Content-Type: text/markdown

# authpi-admin

Official Python Admin SDK for the AuthPI Core API.

**Requirements:** Python 3.11+, async-only (`httpx` + `asyncio`)

## Installation

```bash
pip install authpi-admin
```

## Quick Start

```python
from authpi_admin import AuthPIAdmin

async with AuthPIAdmin(api_key="key_xxx") as admin:
    # List issuers
    page = await admin.issuers.list(limit=10)
    print(page.data)

    # Scope into an issuer and manage users
    users = await admin.issuer("iss_xxx").users.list()

    # Auto-paginate
    async for user in admin.issuer("iss_xxx").users.list_all():
        print(user)

    # Create a user
    user = await admin.issuer("iss_xxx").users.create({
        "email": "alice@example.com",
        "display_name": "Alice",
    })
```

## Authentication

### API Key (default)

```python
admin = AuthPIAdmin(api_key="key_xxx")
```

### Bearer Token

For server-side applications authenticating on behalf of a user session:

```python
admin = AuthPIAdmin(
    access_token="tok_xxx",
    account_id="acc_xxx",
)
```

With optional token refresh callback:

```python
async def refresh():
    new_tokens = await my_refresh_logic()
    return {"access_token": new_tokens.access_token}

admin = AuthPIAdmin(
    access_token="tok_xxx",
    account_id="acc_xxx",
    on_token_expired=refresh,
)
```

When `on_token_expired` is provided, the SDK calls it on 401 responses and retries the request with the new token. Concurrent 401s are deduplicated — only one refresh runs at a time.

## Scoped Client Pattern

The SDK mirrors the API's resource hierarchy. Navigate with chained accessors:

```python
# Account-level resources
await admin.issuers.list()
await admin.webhooks.create({"url": "https://..."})
await admin.events.list(limit=50)

# Issuer scope
iss = admin.issuer("iss_xxx")
await iss.users.list()
await iss.agents.create({"name": "bot"})
await iss.clients.list()
await iss.organizations.list()

# User scope (nested under issuer)
usr = admin.issuer("iss_xxx").user("usr_xxx")
await usr.get()
await usr.sessions.list()
await usr.tokens.list()
await usr.trusted_devices.list()
await usr.verifiers.list()

# Webhook scope
wh = admin.webhook("wh_xxx")
await wh.get()
await wh.deliveries.list()
```

## Pagination

List endpoints return a `Page` with cursor-based pagination:

```python
# Manual pagination
page = await admin.issuer("iss_xxx").users.list(limit=25)
print(page.data)        # list[dict]
print(page.has_more)    # bool
print(page.next_cursor) # str | None

# Fetch next page
if page.has_more:
    next_page = await admin.issuer("iss_xxx").users.list(
        limit=25, cursor=page.next_cursor
    )

# Auto-pagination (yields individual items across all pages)
async for user in admin.issuer("iss_xxx").users.list_all():
    print(user)
```

## Retries

Read-only requests (GET, HEAD, OPTIONS) are automatically retried on 429, 502, 503, and 504 responses with exponential backoff. The `Retry-After` header is respected when present.

```python
# Default: retries enabled (3 attempts, 1s base delay, exponential backoff)
admin = AuthPIAdmin(api_key="key_xxx")

# Disable retries
admin = AuthPIAdmin(api_key="key_xxx", retries=False)

# Custom retry config
admin = AuthPIAdmin(
    api_key="key_xxx",
    retries={"limit": 5, "delay": 0.5, "backoff": "linear"},
)
```

Mutations (POST, PATCH, DELETE) are never retried automatically. Use idempotency keys and handle retries explicitly for writes.

## ETags & Optimistic Concurrency

GET responses include an `_etag` field. Pass it back on updates to prevent overwriting concurrent changes:

```python
user = await admin.issuer("iss_xxx").user("usr_xxx").get()

# Conditional update — fails with PreconditionFailedError if modified
await admin.issuer("iss_xxx").users.update(
    "usr_xxx",
    {"display_name": "Bob"},
    if_match=user.get("_etag"),
)
```

## Error Handling

The SDK maps HTTP status codes to specific error classes:

```python
from authpi_admin import (
    NotFoundError,
    ValidationError,
    AuthenticationError,
    RateLimitError,
    PreconditionFailedError,
)

try:
    await admin.issuer("iss_xxx").user("usr_xxx").get()
except NotFoundError:
    print("User not found")
except ValidationError as err:
    print("Validation failed:", err.fields)
except RateLimitError as err:
    print(f"Retry after {err.retry_after} seconds")
except AuthenticationError:
    print("Invalid credentials")
```

### Error Hierarchy

| Error | Status | Extra Fields | Retryable |
|-------|--------|--------------|-----------|
| `ApiError` | — | `error`, `error_description`, `status_code`, `retryable`, `reference`, `raw_body` | — |
| `ValidationError` | 400, 422 | `fields` | No |
| `AuthenticationError` | 401 | — | No |
| `ForbiddenError` | 403 | — | No |
| `NotFoundError` | 404 | — | No |
| `ConflictError` | 409 | — | No |
| `PreconditionFailedError` | 412 | `current_etag` | No |
| `RateLimitError` | 429 | `retry_after` | Yes |
| `InternalServerError` | 500 | — | No |
| `BadGatewayError` | 502 | — | Yes |
| `ServiceUnavailableError` | 503 | — | Yes |
| `GatewayTimeoutError` | 504 | — | Yes |
| `UnexpectedError` | other | — | No |
| `ClosedClientError` | — | — | No |

## Configuration

```python
from authpi_admin import AuthPIAdmin

admin = AuthPIAdmin(
    api_key="key_xxx",                          # or access_token + account_id
    base_url="https://api.authpi.dev",          # default
    timeout=30.0,                               # default, in seconds
    default_headers={"X-Custom": "value"},      # optional extra headers
    retries=True,                               # default (or False, or dict)
)
```

### Custom httpx Client

Inject a pre-configured `httpx.AsyncClient` for advanced use cases (proxies, certificates, connection pooling):

```python
import httpx
from authpi_admin import AuthPIAdmin

async with httpx.AsyncClient(proxies="http://proxy:8080") as http:
    admin = AuthPIAdmin(api_key="key_xxx", http_client=http)
    await admin.issuers.list()
```

### Context Manager

The SDK supports async context managers for clean resource cleanup:

```python
async with AuthPIAdmin(api_key="key_xxx") as admin:
    await admin.issuers.list()
# httpx client is closed automatically
```

Or close manually:

```python
admin = AuthPIAdmin(api_key="key_xxx")
try:
    await admin.issuers.list()
finally:
    await admin.close()
```

## License

MIT
