authpi-admin 0.3.0

Official Python Admin SDK for AuthPI Core API

authpi-admin

Official Python Admin SDK for the AuthPI Core API.

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

Installation

pip install authpi-admin

Quick Start

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": "[e-mail]",
        "display_name": "Alice",
    })

Authentication

API Key (default)

admin = AuthPIAdmin(api_key="key_xxx")

Bearer Token

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

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

With optional token refresh callback:

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:

# 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:

# 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.

# 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:

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:

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

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):

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:

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

Or close manually:

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

License

MIT