app.services.apikey_service module

class app.services.apikey_service.ApiKeyService(apikey_repository, cache_backend=None)[source]

Bases: BaseService

Service class for API keys.

Plaintext keys are surfaced to the caller exactly once at creation and never persisted. Authentication compares the sha256 hash of the value presented in the X-API-Key header against the canonical apiKeyHash column.

Revocation consistency depends on configs.APIKEY_CACHE_BACKEND:

  • redis (recommended for multi-worker deployments): cache entries live in a shared Redis keyspace, so a revoke on any worker is observed by every other worker on its next request.

  • memory (default, single process): each gunicorn worker keeps its own dict; revocations only invalidate the local worker’s entry and remote workers continue serving the cached value until the TTL (API_KEY_HEADER_CACHE_TTL_SECONDS, default 5s) expires.

Parameters:
apikey_repository

Repository instance for API keys.

Type:

ApiKeyRepository

cache_backend

Backing store for the resolved (apiKey, active) tuple keyed by sha256 hash.

Type:

ApiKeyCacheBackend

__init__(apikey_repository, cache_backend=None)[source]

Initializes the ApiKeyService with the provided repository and cache.

Parameters:
  • apikey_repository (ApiKeyRepository) – The repository instance.

  • cache_backend (ApiKeyCacheBackend | None) – The cache backend. Defaults to a fresh in-process instance so direct instantiation (mostly tests) keeps working without wiring through the DI container.

Return type:

None

async generate_api_key_service()[source]

Generate a cryptographically-secure API key whose prefix and hash do not collide with any existing record.

Return type:

GeneratedApiKey

async create_api_key(apikeyPostBody)[source]

Persist a new API-key record.

Parameters:

apikeyPostBody – Schema describing the API key to store (including its prefix and hash).

Returns:

Any – The created API-key entity.

Return type:

Any

async get_all_api_keys()[source]

Return every API-key record.

Returns:

Any – All stored API-key entities.

Return type:

Any

async revoke_api_key_by_prefix(prefix)[source]

Revoke an API key identified by its public prefix and drop the matching cache entry. The deactivated row’s apiKeyHash is the exact cache key used by get_api_key_header (both are hash_api_key(plaintext)), so a precise delete is enough – no need to nuke unrelated entries.

Parameters:

prefix (str)

Return type:

Any

async static get_api_key_header(api_key=Security(APIKeyHeader))[source]

Authenticate a request based on the value of the X-API-Key header. The value is hashed and compared against the canonical apiKeyHash column; the plaintext is never stored or returned.

A short-lived cache (when API_KEY_HEADER_CACHE_TTL_SECONDS > 0) avoids a DB lookup on every request.

Parameters:

api_key (str) – The raw X-API-Key header value.

Returns:

ResponseResponse.ok with the normalized key info on success, or Response.fail when the header is absent.

Raises:

ForbiddenError – If the key is missing, unknown or inactive.

Return type:

Response

classmethod clear_header_cache()[source]

Reset the cache synchronously. Only the in-memory backend can be cleared without an event loop; for the Redis backend, prefer the async await backend.clear() (or rely on per-entry deletion via revoke_api_key_by_prefix). Kept as a classmethod so existing tests can call it from setUp for isolation.

Return type:

None