Auth
leadr.auth¶
Modules:
- adapters –
- api –
- bootstrap – Superadmin bootstrap functionality.
- dependencies – Authentication dependencies for FastAPI.
- domain –
- services –
leadr.auth.adapters¶
Modules:
- orm – Auth ORM models.
leadr.auth.adapters.orm¶
Auth ORM models.
Classes:
- APIKeyORM – API Key ORM model.
- APIKeyStatusEnum – API Key status enum for database.
- DeviceORM – Device ORM model.
- DeviceSessionORM – DeviceSession ORM model.
- DeviceStatusEnum – Device status enum for database.
- NonceORM – Nonce ORM model.
- NonceStatusEnum – Nonce status enum for database.
leadr.auth.adapters.orm.APIKeyORM¶
Bases: Base
API Key ORM model.
Represents an API key for account authentication in the database. Maps to the api_keys table with foreign key to accounts and users. Each API key is owned by a specific user within the account.
Attributes:
- account_id (
Mapped[UUID]) – - created_at (
Mapped[timestamp]) – - deleted_at (
Mapped[nullable_timestamp]) – - expires_at (
Mapped[datetime | None]) – - id (
Mapped[uuid_pk]) – - key_hash (
Mapped[str]) – - key_prefix (
Mapped[str]) – - last_used_at (
Mapped[datetime | None]) – - name (
Mapped[str]) – - status (
Mapped[APIKeyStatusEnum]) – - updated_at (
Mapped[timestamp]) – - user_id (
Mapped[UUID]) –
# leadr.auth.adapters.orm.APIKeyORM.account_id¶
account_id: Mapped[UUID] = mapped_column(ForeignKey('accounts.id', ondelete='CASCADE'), nullable=False, index=True)
# leadr.auth.adapters.orm.APIKeyORM.created_at¶
created_at: Mapped[timestamp]
# leadr.auth.adapters.orm.APIKeyORM.deleted_at¶
deleted_at: Mapped[nullable_timestamp]
# leadr.auth.adapters.orm.APIKeyORM.expires_at¶
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
# leadr.auth.adapters.orm.APIKeyORM.id¶
id: Mapped[uuid_pk]
# leadr.auth.adapters.orm.APIKeyORM.key_hash¶
key_hash: Mapped[str] = mapped_column(String, nullable=False)
# leadr.auth.adapters.orm.APIKeyORM.key_prefix¶
key_prefix: Mapped[str] = mapped_column(String, nullable=False, unique=True, index=True)
# leadr.auth.adapters.orm.APIKeyORM.last_used_at¶
last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
# leadr.auth.adapters.orm.APIKeyORM.name¶
name: Mapped[str] = mapped_column(String, nullable=False)
# leadr.auth.adapters.orm.APIKeyORM.status¶
status: Mapped[APIKeyStatusEnum] = mapped_column(Enum(APIKeyStatusEnum, name='api_key_status', native_enum=True, values_callable=(lambda x: [(e.value) for e in x])), nullable=False, default=(APIKeyStatusEnum.ACTIVE), server_default='active')
# leadr.auth.adapters.orm.APIKeyORM.updated_at¶
updated_at: Mapped[timestamp] = mapped_column(onupdate=(func.now()))
# leadr.auth.adapters.orm.APIKeyORM.user_id¶
user_id: Mapped[UUID] = mapped_column(ForeignKey('users.id', ondelete='CASCADE'), nullable=False, index=True)
leadr.auth.adapters.orm.APIKeyStatusEnum¶
API Key status enum for database.
Attributes:
# leadr.auth.adapters.orm.APIKeyStatusEnum.ACTIVE¶
ACTIVE = 'active'
# leadr.auth.adapters.orm.APIKeyStatusEnum.REVOKED¶
REVOKED = 'revoked'
leadr.auth.adapters.orm.DeviceORM¶
Bases: Base
Device ORM model.
Represents a game client device (e.g., mobile device, PC, console) in the database. Maps to the devices table with foreign key to games. Devices are scoped per-game for client authentication.
Functions:
- from_domain – Convert Device domain entity to ORM model.
- to_domain – Convert ORM model to Device domain entity.
Attributes:
- account_id (
Mapped[UUID]) – - client_fingerprint (
Mapped[str]) – - created_at (
Mapped[timestamp]) – - deleted_at (
Mapped[nullable_timestamp]) – - device_metadata (
Mapped[dict[str, Any]]) – - first_seen_at (
Mapped[datetime]) – - game (
Mapped[GameORM]) – - game_id (
Mapped[UUID]) – - id (
Mapped[uuid_pk]) – - last_seen_at (
Mapped[datetime]) – - nonces (
Mapped[list[NonceORM]]) – - platform (
Mapped[str | None]) – - sessions (
Mapped[list[DeviceSessionORM]]) – - status (
Mapped[DeviceStatusEnum]) – - updated_at (
Mapped[timestamp]) –
# leadr.auth.adapters.orm.DeviceORM.account_id¶
account_id: Mapped[UUID] = mapped_column(nullable=False, index=True)
# leadr.auth.adapters.orm.DeviceORM.client_fingerprint¶
client_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False)
# leadr.auth.adapters.orm.DeviceORM.created_at¶
created_at: Mapped[timestamp]
# leadr.auth.adapters.orm.DeviceORM.deleted_at¶
deleted_at: Mapped[nullable_timestamp]
# leadr.auth.adapters.orm.DeviceORM.device_metadata¶
device_metadata: Mapped[dict[str, Any]] = mapped_column('metadata', JSON, nullable=False, default=dict, server_default='{}')
# leadr.auth.adapters.orm.DeviceORM.first_seen_at¶
first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
# leadr.auth.adapters.orm.DeviceORM.from_domain¶
from_domain(device)
Convert Device domain entity to ORM model.
# leadr.auth.adapters.orm.DeviceORM.game¶
game: Mapped[GameORM] = relationship('GameORM')
# leadr.auth.adapters.orm.DeviceORM.game_id¶
game_id: Mapped[UUID] = mapped_column(ForeignKey('games.id', ondelete='CASCADE'), nullable=False, index=True)
# leadr.auth.adapters.orm.DeviceORM.id¶
id: Mapped[uuid_pk]
# leadr.auth.adapters.orm.DeviceORM.last_seen_at¶
last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
# leadr.auth.adapters.orm.DeviceORM.nonces¶
nonces: Mapped[list[NonceORM]] = relationship('NonceORM', cascade='all, delete-orphan')
# leadr.auth.adapters.orm.DeviceORM.platform¶
platform: Mapped[str | None] = mapped_column(String, nullable=True)
# leadr.auth.adapters.orm.DeviceORM.sessions¶
sessions: Mapped[list[DeviceSessionORM]] = relationship('DeviceSessionORM', back_populates='device', cascade='all, delete-orphan')
# leadr.auth.adapters.orm.DeviceORM.status¶
status: Mapped[DeviceStatusEnum] = mapped_column(Enum(DeviceStatusEnum, name='device_status', native_enum=True, values_callable=(lambda x: [(e.value) for e in x])), nullable=False, default=(DeviceStatusEnum.ACTIVE), server_default='active')
# leadr.auth.adapters.orm.DeviceORM.to_domain¶
to_domain()
Convert ORM model to Device domain entity.
# leadr.auth.adapters.orm.DeviceORM.updated_at¶
updated_at: Mapped[timestamp] = mapped_column(onupdate=(func.now()))
leadr.auth.adapters.orm.DeviceSessionORM¶
Bases: Base
DeviceSession ORM model.
Represents an active authentication session for a device in the database. Maps to the device_sessions table with foreign key to devices. Sessions include both access and refresh tokens with token rotation support.
Functions:
- from_domain – Convert DeviceSession domain entity to ORM model.
- to_domain – Convert ORM model to DeviceSession domain entity.
Attributes:
- access_token_hash (
Mapped[str]) – - created_at (
Mapped[timestamp]) – - deleted_at (
Mapped[nullable_timestamp]) – - device (
Mapped[DeviceORM]) – - device_id (
Mapped[UUID]) – - expires_at (
Mapped[datetime]) – - id (
Mapped[uuid_pk]) – - ip_address (
Mapped[str | None]) – - refresh_expires_at (
Mapped[datetime]) – - refresh_token_hash (
Mapped[str]) – - revoked_at (
Mapped[datetime | None]) – - token_version (
Mapped[int]) – - updated_at (
Mapped[timestamp]) – - user_agent (
Mapped[str | None]) –
# leadr.auth.adapters.orm.DeviceSessionORM.access_token_hash¶
access_token_hash: Mapped[str] = mapped_column(String, nullable=False, index=True)
# leadr.auth.adapters.orm.DeviceSessionORM.created_at¶
created_at: Mapped[timestamp]
# leadr.auth.adapters.orm.DeviceSessionORM.deleted_at¶
deleted_at: Mapped[nullable_timestamp]
# leadr.auth.adapters.orm.DeviceSessionORM.device¶
device: Mapped[DeviceORM] = relationship('DeviceORM', back_populates='sessions')
# leadr.auth.adapters.orm.DeviceSessionORM.device_id¶
device_id: Mapped[UUID] = mapped_column(ForeignKey('devices.id', ondelete='CASCADE'), nullable=False, index=True)
# leadr.auth.adapters.orm.DeviceSessionORM.expires_at¶
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)
# leadr.auth.adapters.orm.DeviceSessionORM.from_domain¶
from_domain(session)
Convert DeviceSession domain entity to ORM model.
# leadr.auth.adapters.orm.DeviceSessionORM.id¶
id: Mapped[uuid_pk]
# leadr.auth.adapters.orm.DeviceSessionORM.ip_address¶
ip_address: Mapped[str | None] = mapped_column(String, nullable=True)
# leadr.auth.adapters.orm.DeviceSessionORM.refresh_expires_at¶
refresh_expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)
# leadr.auth.adapters.orm.DeviceSessionORM.refresh_token_hash¶
refresh_token_hash: Mapped[str] = mapped_column(String, nullable=False, index=True)
# leadr.auth.adapters.orm.DeviceSessionORM.revoked_at¶
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
# leadr.auth.adapters.orm.DeviceSessionORM.to_domain¶
to_domain()
Convert ORM model to DeviceSession domain entity.
# leadr.auth.adapters.orm.DeviceSessionORM.token_version¶
token_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1, server_default='1')
# leadr.auth.adapters.orm.DeviceSessionORM.updated_at¶
updated_at: Mapped[timestamp] = mapped_column(onupdate=(func.now()))
# leadr.auth.adapters.orm.DeviceSessionORM.user_agent¶
user_agent: Mapped[str | None] = mapped_column(String, nullable=True)
leadr.auth.adapters.orm.DeviceStatusEnum¶
Device status enum for database.
Attributes:
# leadr.auth.adapters.orm.DeviceStatusEnum.ACTIVE¶
ACTIVE = 'active'
# leadr.auth.adapters.orm.DeviceStatusEnum.BANNED¶
BANNED = 'banned'
# leadr.auth.adapters.orm.DeviceStatusEnum.SUSPENDED¶
SUSPENDED = 'suspended'
leadr.auth.adapters.orm.NonceORM¶
Bases: Base
Nonce ORM model.
Represents a single-use nonce for replay protection in the database. Maps to the nonces table with foreign key to devices. Nonces are short-lived tokens (typically 60 seconds) that must be obtained before making mutating requests.
Functions:
- from_domain – Convert Nonce domain entity to ORM model.
- to_domain – Convert ORM model to Nonce domain entity.
Attributes:
- created_at (
Mapped[timestamp]) – - deleted_at (
Mapped[nullable_timestamp]) – - device (
Mapped[DeviceORM]) – - device_id (
Mapped[UUID]) – - expires_at (
Mapped[datetime]) – - id (
Mapped[uuid_pk]) – - nonce_value (
Mapped[str]) – - status (
Mapped[NonceStatusEnum]) – - updated_at (
Mapped[timestamp]) – - used_at (
Mapped[datetime | None]) –
# leadr.auth.adapters.orm.NonceORM.created_at¶
created_at: Mapped[timestamp]
# leadr.auth.adapters.orm.NonceORM.deleted_at¶
deleted_at: Mapped[nullable_timestamp]
# leadr.auth.adapters.orm.NonceORM.device¶
device: Mapped[DeviceORM] = relationship('DeviceORM', overlaps='nonces')
# leadr.auth.adapters.orm.NonceORM.device_id¶
device_id: Mapped[UUID] = mapped_column(ForeignKey('devices.id', ondelete='CASCADE'), nullable=False, index=True)
# leadr.auth.adapters.orm.NonceORM.expires_at¶
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)
# leadr.auth.adapters.orm.NonceORM.from_domain¶
from_domain(nonce)
Convert Nonce domain entity to ORM model.
# leadr.auth.adapters.orm.NonceORM.id¶
id: Mapped[uuid_pk]
# leadr.auth.adapters.orm.NonceORM.nonce_value¶
nonce_value: Mapped[str] = mapped_column(String, nullable=False, unique=True, index=True)
# leadr.auth.adapters.orm.NonceORM.status¶
status: Mapped[NonceStatusEnum] = mapped_column(Enum(NonceStatusEnum, name='nonce_status', native_enum=True, values_callable=(lambda x: [(e.value) for e in x])), nullable=False, default=(NonceStatusEnum.PENDING), server_default='pending')
# leadr.auth.adapters.orm.NonceORM.to_domain¶
to_domain()
Convert ORM model to Nonce domain entity.
# leadr.auth.adapters.orm.NonceORM.updated_at¶
updated_at: Mapped[timestamp] = mapped_column(onupdate=(func.now()))
# leadr.auth.adapters.orm.NonceORM.used_at¶
used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
leadr.auth.adapters.orm.NonceStatusEnum¶
Nonce status enum for database.
Attributes:
# leadr.auth.adapters.orm.NonceStatusEnum.EXPIRED¶
EXPIRED = 'expired'
# leadr.auth.adapters.orm.NonceStatusEnum.PENDING¶
PENDING = 'pending'
# leadr.auth.adapters.orm.NonceStatusEnum.USED¶
USED = 'used'
leadr.auth.api¶
Modules:
- api_key_routes – API routes for authentication and API key management.
- api_key_schemas – API schemas for Authentication endpoints.
- client_routes – API routes for client device authentication.
- client_schemas – API request and response models for client authentication.
- device_routes – API routes for device management.
- device_schemas – API request and response models for devices.
- device_session_routes – API routes for device session management.
- device_session_schemas – API schemas for device sessions.
leadr.auth.api.api_key_routes¶
API routes for authentication and API key management.
Functions:
- create_api_key – Create a new API key for an account.
- get_api_key – Get a single API key by ID.
- list_api_keys – List API keys for an account with optional filters and pagination.
- update_api_key – Update an API key.
Attributes:
- router –
leadr.auth.api.api_key_routes.create_api_key¶
create_api_key(request, service, auth)
Create a new API key for an account.
The plain API key is returned only once in this response. Store it securely as it cannot be retrieved later.
For regular users, account_id must match their API key's account. For superadmins, any account_id is accepted.
Returns:
CreateAPIKeyResponse– CreateAPIKeyResponse with the plain key included.
Raises:
403– User does not have access to the specified account.404– Account not found.
leadr.auth.api.api_key_routes.get_api_key¶
get_api_key(key_id, service, auth)
Get a single API key by ID.
Parameters:
- key_id (
APIKeyID) – The UUID of the API key to retrieve. - service (
APIKeyServiceDep) – Injected API key service dependency. - auth (
AdminAuthContextDep) – Authentication context with user info.
Returns:
APIKeyResponse– APIKeyResponse with key details (excludes the plain key).
Raises:
403– User does not have access to this API key's account.404– API key not found.
leadr.auth.api.api_key_routes.list_api_keys¶
list_api_keys(auth, service, pagination, account_id=None, key_status=None)
List API keys for an account with optional filters and pagination.
For regular users, account_id is automatically derived from their API key. For superadmins, account_id is optional - if omitted, returns API keys from all accounts.
Pagination:
- Default: 20 items per page, sorted by created_at:desc,id:asc
- Custom sort: Use ?sort=name:asc,created_at:desc
- Valid sort fields: id, name, created_at, updated_at
- Navigation: Use next_cursor/prev_cursor from response
Example
GET /v1/api-keys?account_id=acc_123&status=active&limit=50&sort=name:ascParameters:
- auth (
AdminAuthContextDep) – Authentication context with user info. - service (
APIKeyServiceDep) – Injected API key service dependency. - pagination (
Annotated[PaginationParams, Depends()]) – Pagination parameters (cursor, limit, sort). - account_id (
Annotated[AccountID | None, Query(description='Account ID filter')]) – Optional account_id query parameter (superadmins can omit to see all). - key_status (
Annotated[APIKeyStatus | None, Query(alias=status, description='Filter by status')]) – Optional status to filter results (active or revoked).
Returns:
PaginatedResponse[APIKeyResponse]– PaginatedResponse with API keys and pagination metadata.
Raises:
400– Invalid cursor, sort field, or cursor state mismatch.403– User does not have access to the specified account.
leadr.auth.api.api_key_routes.router¶
router = APIRouter()
leadr.auth.api.api_key_routes.update_api_key¶
update_api_key(key_id, request, service, auth)
Update an API key.
Currently supports:
- Updating status (e.g., to revoke a key)
- Soft delete via deleted flag
Parameters:
- key_id (
APIKeyID) – The UUID of the API key to update. - request (
UpdateAPIKeyRequest) – Update request with optional status and deleted fields. - service (
APIKeyServiceDep) – Injected API key service dependency. - auth (
AdminAuthContextDep) – Authentication context with user info.
Returns:
APIKeyResponse– APIKeyResponse with updated key details.
Raises:
403– User does not have access to this API key's account.404– API key not found.
leadr.auth.api.api_key_schemas¶
API schemas for Authentication endpoints.
Classes:
- APIKeyResponse – Response schema for API key details.
- CreateAPIKeyRequest – Request schema for creating an API key.
- CreateAPIKeyResponse – Response schema for creating an API key.
- UpdateAPIKeyRequest – Request schema for updating an API key.
leadr.auth.api.api_key_schemas.APIKeyResponse¶
Bases: BaseModel
Response schema for API key details.
Excludes sensitive information like key_hash. The full key is never returned after creation.
Functions:
- from_domain – Convert domain entity to response model.
Attributes:
- account_id (
AccountID) – - created_at (
datetime) – - expires_at (
datetime | None) – - id (
APIKeyID) – - last_used_at (
datetime | None) – - model_config –
- name (
str) – - prefix (
str) – - status (
APIKeyStatus) – - updated_at (
datetime) – - user_id (
UserID) –
# leadr.auth.api.api_key_schemas.APIKeyResponse.account_id¶
account_id: AccountID = Field(description='ID of the account this key belongs to')
# leadr.auth.api.api_key_schemas.APIKeyResponse.created_at¶
created_at: datetime = Field(description='Timestamp when the key was created (UTC)')
# leadr.auth.api.api_key_schemas.APIKeyResponse.expires_at¶
expires_at: datetime | None = Field(default=None, description='Expiration timestamp (UTC), or null if never expires')
# leadr.auth.api.api_key_schemas.APIKeyResponse.from_domain¶
from_domain(api_key)
Convert domain entity to response model.
Parameters:
- api_key (
APIKey) – The domain APIKey entity
Returns:
APIKeyResponse– APIKeyResponse with all fields populated
# leadr.auth.api.api_key_schemas.APIKeyResponse.id¶
id: APIKeyID = Field(description='Unique identifier for the API key')
# leadr.auth.api.api_key_schemas.APIKeyResponse.last_used_at¶
last_used_at: datetime | None = Field(default=None, description='Timestamp of last successful authentication (UTC)')
# leadr.auth.api.api_key_schemas.APIKeyResponse.model_config¶
model_config = ConfigDict(extra='forbid')
# leadr.auth.api.api_key_schemas.APIKeyResponse.name¶
name: str = Field(description='Human-readable name for the API key')
# leadr.auth.api.api_key_schemas.APIKeyResponse.prefix¶
prefix: str = Field(description='Key prefix for identification (first 8 characters)')
# leadr.auth.api.api_key_schemas.APIKeyResponse.status¶
status: APIKeyStatus = Field(description='Current status (active, revoked, expired)')
# leadr.auth.api.api_key_schemas.APIKeyResponse.updated_at¶
updated_at: datetime = Field(description='Timestamp of last update (UTC)')
# leadr.auth.api.api_key_schemas.APIKeyResponse.user_id¶
user_id: UserID = Field(description='ID of the user who owns this API key')
leadr.auth.api.api_key_schemas.CreateAPIKeyRequest¶
Bases: BaseModel
Request schema for creating an API key.
Attributes:
- account_id (
AccountID) – - expires_at (
datetime | None) – - name (
str) – - user_id (
UserID) –
# leadr.auth.api.api_key_schemas.CreateAPIKeyRequest.account_id¶
account_id: AccountID = Field(description='ID of the account this API key belongs to')
# leadr.auth.api.api_key_schemas.CreateAPIKeyRequest.expires_at¶
expires_at: datetime | None = Field(default=None, description='Optional expiration timestamp (UTC). Key never expires if omitted')
# leadr.auth.api.api_key_schemas.CreateAPIKeyRequest.name¶
name: str = Field(description="Human-readable name for the API key (e.g., 'Production Server')")
# leadr.auth.api.api_key_schemas.CreateAPIKeyRequest.user_id¶
user_id: UserID = Field(description='ID of the user who owns this API key')
leadr.auth.api.api_key_schemas.CreateAPIKeyResponse¶
Bases: BaseModel
Response schema for creating an API key.
Includes the plain API key which is only shown once. The client must save this key as it cannot be retrieved later.
Functions:
- from_domain – Convert domain entity to response model with plain key.
Attributes:
- created_at (
datetime) – - expires_at (
datetime | None) – - id (
APIKeyID) – - key (
str) – - name (
str) – - prefix (
str) – - status (
APIKeyStatus) –
# leadr.auth.api.api_key_schemas.CreateAPIKeyResponse.created_at¶
created_at: datetime = Field(description='Timestamp when the key was created (UTC)')
# leadr.auth.api.api_key_schemas.CreateAPIKeyResponse.expires_at¶
expires_at: datetime | None = Field(default=None, description='Expiration timestamp (UTC), or null if never expires')
# leadr.auth.api.api_key_schemas.CreateAPIKeyResponse.from_domain¶
from_domain(api_key, plain_key)
Convert domain entity to response model with plain key.
Parameters:
- api_key (
APIKey) – The domain APIKey entity - plain_key (
str) – The plain text API key (only available at creation)
Returns:
CreateAPIKeyResponse– CreateAPIKeyResponse with all fields populated
# leadr.auth.api.api_key_schemas.CreateAPIKeyResponse.id¶
id: APIKeyID = Field(description='Unique identifier for the API key')
# leadr.auth.api.api_key_schemas.CreateAPIKeyResponse.key¶
key: str = Field(description='Plain text API key. ONLY returned at creation - save this securely!')
# leadr.auth.api.api_key_schemas.CreateAPIKeyResponse.name¶
name: str = Field(description='Human-readable name for the API key')
# leadr.auth.api.api_key_schemas.CreateAPIKeyResponse.prefix¶
prefix: str = Field(description='Key prefix for identification (first 8 characters)')
# leadr.auth.api.api_key_schemas.CreateAPIKeyResponse.status¶
status: APIKeyStatus = Field(description='Current status of the API key')
leadr.auth.api.api_key_schemas.UpdateAPIKeyRequest¶
Bases: BaseModel
Request schema for updating an API key.
Can update status (e.g., to revoke) or set deleted flag for soft delete.
Attributes:
- deleted (
bool | None) – - status (
APIKeyStatus | None) –
# leadr.auth.api.api_key_schemas.UpdateAPIKeyRequest.deleted¶
deleted: bool | None = Field(default=None, description='Set to true to soft delete the key')
# leadr.auth.api.api_key_schemas.UpdateAPIKeyRequest.status¶
status: APIKeyStatus | None = Field(default=None, description="Updated status (use 'revoked' to disable key)")
leadr.auth.api.client_routes¶
API routes for client device authentication.
Functions:
- generate_nonce – Generate a fresh nonce for replay protection.
- refresh_session – Refresh an expired access token using a valid refresh token.
- start_session – Start a new device session for a game client.
Attributes:
leadr.auth.api.client_routes.generate_nonce¶
generate_nonce(auth, service)
Generate a fresh nonce for replay protection.
Nonces are single-use tokens with short TTL (60 seconds) that clients must obtain before making mutating requests (POST, PATCH, DELETE). This prevents replay attacks by ensuring each request is fresh and authorized.
Requires device authentication via access token.
Parameters:
- auth (
ClientAuthContextDep) – Authenticated client auth context (device guaranteed non-None) - service (
NonceServiceDep) – NonceService dependency
Returns:
NonceResponse– NonceResponse with nonce_value and expires_at
Raises:
401– Invalid or missing device token
Example
1. Client calls GET /client/nonce with Authorization header 1. Server returns nonce_value and expires_at 1. Client includes nonce in leadr-client-nonce header for mutations 1. Server validates and consumes nonce (single-use)leadr.auth.api.client_routes.protected_router¶
protected_router = APIRouter()
leadr.auth.api.client_routes.public_router¶
public_router = APIRouter(prefix='/client')
leadr.auth.api.client_routes.refresh_session¶
refresh_session(request, service)
Refresh an expired access token using a valid refresh token.
This endpoint implements token rotation for security:
- Returns new access and refresh tokens
- Increments the token version
- Invalidates the old refresh token (prevents replay attacks)
No authentication is required (the refresh token itself is the credential).
Parameters:
- request (
RefreshTokenRequest) – Refresh token request - service (
DeviceServiceDep) – DeviceService dependency
Returns:
RefreshTokenResponse– RefreshTokenResponse with new tokens
Raises:
401– Invalid or expired refresh token422– Invalid request (missing refresh_token)
leadr.auth.api.client_routes.start_session¶
start_session(request, service)
Start a new device session for a game client.
This endpoint authenticates game clients and provides JWT access tokens. It is idempotent - calling multiple times for the same device updates last_seen_at and generates a new access token.
No authentication is required to call this endpoint (it IS the authentication).
Parameters:
- request (
StartSessionRequest) – Session start request with game_id and device_id - service (
DeviceServiceDep) – DeviceService dependency
Returns:
StartSessionResponse– StartSessionResponse with device info and access token
Raises:
404– Game not found422– Invalid request (missing required fields, invalid UUID format)
leadr.auth.api.client_schemas¶
API request and response models for client authentication.
Classes:
- NonceResponse – Response schema for nonce generation.
- RefreshTokenRequest – Request schema for refreshing an access token.
- RefreshTokenResponse – Response schema for token refresh.
- StartSessionRequest – Request schema for starting a device session.
- StartSessionResponse – Response schema for starting a device session.
leadr.auth.api.client_schemas.NonceResponse¶
Bases: BaseModel
Response schema for nonce generation.
Nonces are single-use tokens with short TTL (60 seconds) that clients must obtain before making mutating requests. This prevents replay attacks.
Attributes:
- expires_at (
datetime) – - nonce_value (
str) –
# leadr.auth.api.client_schemas.NonceResponse.expires_at¶
expires_at: datetime = Field(description='Nonce expiration timestamp (UTC)')
# leadr.auth.api.client_schemas.NonceResponse.nonce_value¶
nonce_value: str = Field(description='Unique nonce value (UUID)')
leadr.auth.api.client_schemas.RefreshTokenRequest¶
Bases: BaseModel
Request schema for refreshing an access token.
Used by clients when their access token has expired.
Attributes:
- refresh_token (
str) –
# leadr.auth.api.client_schemas.RefreshTokenRequest.refresh_token¶
refresh_token: str = Field(description='JWT refresh token obtained from start_session')
leadr.auth.api.client_schemas.RefreshTokenResponse¶
Bases: BaseModel
Response schema for token refresh.
Returns new access and refresh tokens with incremented version. The old refresh token is invalidated and cannot be reused.
Attributes:
- access_token (
str) – - expires_in (
int) – - refresh_token (
str) –
# leadr.auth.api.client_schemas.RefreshTokenResponse.access_token¶
access_token: str = Field(description='New JWT access token')
# leadr.auth.api.client_schemas.RefreshTokenResponse.expires_in¶
expires_in: int = Field(description='Access token expiration time in seconds')
# leadr.auth.api.client_schemas.RefreshTokenResponse.refresh_token¶
refresh_token: str = Field(description='New JWT refresh token (old token is invalidated)')
leadr.auth.api.client_schemas.StartSessionRequest¶
Bases: BaseModel
Request schema for starting a device session.
Used by game clients to authenticate and obtain an access token.
Attributes:
- client_fingerprint (
str) – - game_id (
GameID) – - metadata (
dict[str, Any] | None) – - platform (
str | None) –
# leadr.auth.api.client_schemas.StartSessionRequest.client_fingerprint¶
client_fingerprint: str = Field(description='Client-generated SHA256 device fingerprint (64 hex characters)')
# leadr.auth.api.client_schemas.StartSessionRequest.game_id¶
game_id: GameID = Field(description='ID of the game this device belongs to')
# leadr.auth.api.client_schemas.StartSessionRequest.metadata¶
metadata: dict[str, Any] | None = Field(default=None, description='Optional device metadata (e.g., OS version, device model)')
# leadr.auth.api.client_schemas.StartSessionRequest.platform¶
platform: str | None = Field(default=None, description="Device platform (e.g., 'ios', 'android', 'pc', 'console')")
leadr.auth.api.client_schemas.StartSessionResponse¶
Bases: BaseModel
Response schema for starting a device session.
Includes both access and refresh tokens which must be saved by the client.
- Access token: Short-lived, used for API requests in Authorization header
- Refresh token: Long-lived, used to obtain new access tokens when expired
Functions:
- from_domain – Convert domain entity to response model with tokens.
Attributes:
- access_token (
str) – - account_id (
AccountID) – - client_fingerprint (
str) – - expires_in (
int) – - first_seen_at (
datetime) – - game_id (
GameID) – - id (
DeviceID) – - last_seen_at (
datetime) – - metadata (
dict[str, Any]) – - platform (
str | None) – - refresh_token (
str) – - status (
DeviceStatus) –
# leadr.auth.api.client_schemas.StartSessionResponse.access_token¶
access_token: str = Field(description='JWT access token for authenticating API requests')
# leadr.auth.api.client_schemas.StartSessionResponse.account_id¶
account_id: AccountID = Field(description='ID of the account that owns the game')
# leadr.auth.api.client_schemas.StartSessionResponse.client_fingerprint¶
client_fingerprint: str = Field(description='Client-generated SHA256 device fingerprint (64 hex characters)')
# leadr.auth.api.client_schemas.StartSessionResponse.expires_in¶
expires_in: int = Field(description='Access token expiration time in seconds')
# leadr.auth.api.client_schemas.StartSessionResponse.first_seen_at¶
first_seen_at: datetime = Field(description='Timestamp when device was first seen (UTC)')
# leadr.auth.api.client_schemas.StartSessionResponse.from_domain¶
from_domain(device, access_token, refresh_token, expires_in)
Convert domain entity to response model with tokens.
Parameters:
- device (
Device) – The domain Device entity - access_token (
str) – The plain JWT access token - refresh_token (
str) – The plain JWT refresh token - expires_in (
int) – Access token expiration time in seconds
Returns:
StartSessionResponse– StartSessionResponse with all fields populated
# leadr.auth.api.client_schemas.StartSessionResponse.game_id¶
game_id: GameID = Field(description='ID of the game')
# leadr.auth.api.client_schemas.StartSessionResponse.id¶
id: DeviceID = Field(description='Unique identifier for the device')
# leadr.auth.api.client_schemas.StartSessionResponse.last_seen_at¶
last_seen_at: datetime = Field(description='Timestamp when device was last seen (UTC)')
# leadr.auth.api.client_schemas.StartSessionResponse.metadata¶
metadata: dict[str, Any] = Field(default_factory=dict, description='Device metadata')
# leadr.auth.api.client_schemas.StartSessionResponse.platform¶
platform: str | None = Field(default=None, description='Device platform')
# leadr.auth.api.client_schemas.StartSessionResponse.refresh_token¶
refresh_token: str = Field(description='JWT refresh token for obtaining new access tokens')
# leadr.auth.api.client_schemas.StartSessionResponse.status¶
status: DeviceStatus = Field(description='Device status (active, suspended, banned)')
leadr.auth.api.device_routes¶
API routes for device management.
Functions:
- get_device – Get a device by ID.
- list_devices – List devices for an account with optional filters and pagination.
- update_device – Update a device (change status).
Attributes:
- router –
leadr.auth.api.device_routes.get_device¶
get_device(device_id, service, auth)
Get a device by ID.
Parameters:
- device_id (
DeviceID) – Device identifier to retrieve. - service (
DeviceServiceDep) – Injected device service dependency. - auth (
AdminAuthContextDep) – Authentication context with user info.
Returns:
DeviceResponse– DeviceResponse with the device details.
Raises:
403– User does not have access to this device's account.404– Device not found or soft-deleted.
leadr.auth.api.device_routes.list_devices¶
list_devices(auth, service, pagination, account_id=None, game_id=None, device_status=None)
List devices for an account with optional filters and pagination.
Returns all non-deleted devices for the specified account, with optional filtering by game or status.
For regular users, account_id is automatically derived from their API key. For superadmins, account_id is optional - if omitted, returns devices from all accounts.
Pagination:
- Default: 20 items per page, sorted by created_at:desc,id:asc
- Custom sort: Use ?sort=name:asc,created_at:desc
- Valid sort fields: id, platform, created_at, updated_at
- Navigation: Use next_cursor/prev_cursor from response
Example
GET /v1/devices?account_id=acc_123&game_id=game_456&status=active&limit=50Parameters:
- auth (
AdminAuthContextDep) – Authentication context with user info. - service (
DeviceServiceDep) – Injected device service dependency. - pagination (
Annotated[PaginationParams, Depends()]) – Pagination parameters (cursor, limit, sort). - account_id (
Annotated[AccountID | None, Query(description='Account ID filter')]) – Optional account_id query parameter (superadmins can omit to see all). - game_id (
Annotated[GameID | None, Query(description='Filter by game ID')]) – Optional game ID to filter by. - device_status (
Annotated[str | None, Query(alias=status, description='Filter by status')]) – Optional status to filter by (active, banned, suspended).
Returns:
PaginatedResponse[DeviceResponse]– PaginatedResponse with devices and pagination metadata.
Raises:
400– Invalid cursor, sort field, or cursor state mismatch.403– User does not have access to the specified account.
leadr.auth.api.device_routes.router¶
router = APIRouter()
leadr.auth.api.device_routes.update_device¶
update_device(device_id, request, service, auth)
Update a device (change status).
Allows changing device status to ban, suspend, or activate devices.
Parameters:
- device_id (
DeviceID) – Device identifier to update. - request (
DeviceUpdateRequest) – Update details (status). - service (
DeviceServiceDep) – Injected device service dependency. - auth (
AdminAuthContextDep) – Authentication context with user info.
Returns:
DeviceResponse– DeviceResponse with the updated device details.
Raises:
403– User does not have access to this device's account.404– Device not found.400– Invalid status value.
leadr.auth.api.device_schemas¶
API request and response models for devices.
Classes:
- DeviceResponse – Response model for a device.
- DeviceUpdateRequest – Request model for updating a device.
leadr.auth.api.device_schemas.DeviceResponse¶
Bases: BaseModel
Response model for a device.
Functions:
- from_domain – Convert domain entity to response model.
Attributes:
- account_id (
AccountID) – - client_fingerprint (
str) – - created_at (
datetime) – - first_seen_at (
datetime) – - game_id (
GameID) – - id (
DeviceID) – - last_seen_at (
datetime) – - metadata (
dict[str, Any]) – - platform (
str | None) – - status (
str) – - updated_at (
datetime) –
# leadr.auth.api.device_schemas.DeviceResponse.account_id¶
account_id: AccountID = Field(description='ID of the account this device belongs to')
# leadr.auth.api.device_schemas.DeviceResponse.client_fingerprint¶
client_fingerprint: str = Field(description='Client-generated SHA256 device fingerprint (64 hex characters)')
# leadr.auth.api.device_schemas.DeviceResponse.created_at¶
created_at: datetime = Field(description='Timestamp when device record was created (UTC)')
# leadr.auth.api.device_schemas.DeviceResponse.first_seen_at¶
first_seen_at: datetime = Field(description='Timestamp when device was first seen (UTC)')
# leadr.auth.api.device_schemas.DeviceResponse.from_domain¶
from_domain(device)
Convert domain entity to response model.
Parameters:
- device (
Device) – The domain Device entity to convert.
Returns:
DeviceResponse– DeviceResponse with all fields populated from the domain entity.
# leadr.auth.api.device_schemas.DeviceResponse.game_id¶
game_id: GameID = Field(description='ID of the game this device belongs to')
# leadr.auth.api.device_schemas.DeviceResponse.id¶
id: DeviceID = Field(description='Unique identifier for the device')
# leadr.auth.api.device_schemas.DeviceResponse.last_seen_at¶
last_seen_at: datetime = Field(description='Timestamp when device was last seen (UTC)')
# leadr.auth.api.device_schemas.DeviceResponse.metadata¶
metadata: dict[str, Any] = Field(description='Additional device metadata')
# leadr.auth.api.device_schemas.DeviceResponse.platform¶
platform: str | None = Field(default=None, description='Platform (iOS, Android, etc.), or null')
# leadr.auth.api.device_schemas.DeviceResponse.status¶
status: str = Field(description='Device status: active, banned, or suspended')
# leadr.auth.api.device_schemas.DeviceResponse.updated_at¶
updated_at: datetime = Field(description='Timestamp of last update (UTC)')
leadr.auth.api.device_schemas.DeviceUpdateRequest¶
Bases: BaseModel
Request model for updating a device.
Attributes:
# leadr.auth.api.device_schemas.DeviceUpdateRequest.status¶
status: str | None = Field(default=None, description='Updated status: active, banned, or suspended')
leadr.auth.api.device_session_routes¶
API routes for device session management.
Functions:
- get_session – Get a device session by ID.
- list_sessions – List device sessions for an account with optional filters and pagination.
- update_session – Update a device session (revoke).
Attributes:
- router –
leadr.auth.api.device_session_routes.get_session¶
get_session(session_id, service, auth)
Get a device session by ID.
Parameters:
- session_id (
DeviceSessionID) – Session identifier to retrieve. - service (
DeviceServiceDep) – Injected device service dependency. - auth (
AdminAuthContextDep) – Authentication context with user info.
Returns:
DeviceSessionResponse– DeviceSessionResponse with the session details.
Raises:
403– User does not have access to this session's account.404– Session not found or soft-deleted.
leadr.auth.api.device_session_routes.list_sessions¶
list_sessions(auth, service, pagination, account_id=None, device_id=None)
List device sessions for an account with optional filters and pagination.
Returns all non-deleted device sessions for the specified account, with optional filtering by device.
For regular users, account_id is automatically derived from their API key. For superadmins, account_id is optional - if omitted, returns sessions from all accounts.
Pagination:
- Default: 20 items per page, sorted by created_at:desc,id:asc
- Custom sort: Use ?sort=created_at:asc,id:desc
- Valid sort fields: id, created_at, updated_at
- Navigation: Use next_cursor/prev_cursor from response
Example
GET /v1/device-sessions?account_id=acc_123&device_id=dev_456&limit=50Parameters:
- auth (
AdminAuthContextDep) – Authentication context with user info. - service (
DeviceServiceDep) – Injected device service dependency. - pagination (
Annotated[PaginationParams, Depends()]) – Pagination parameters (cursor, limit, sort). - account_id (
Annotated[AccountID | None, Query(description='Account ID filter')]) – Optional account_id query parameter (superadmins can omit to see all). - device_id (
Annotated[DeviceID | None, Query(description='Filter by device ID')]) – Optional device ID to filter by.
Returns:
PaginatedResponse[DeviceSessionResponse]– PaginatedResponse with device sessions and pagination metadata.
Raises:
400– Invalid cursor, sort field, or cursor state mismatch.403– User does not have access to the specified account.
leadr.auth.api.device_session_routes.router¶
router = APIRouter()
leadr.auth.api.device_session_routes.update_session¶
update_session(session_id, request, service, auth)
Update a device session (revoke).
Allows revoking a device session to invalidate authentication.
Parameters:
- session_id (
DeviceSessionID) – Session identifier to update. - request (
DeviceSessionUpdateRequest) – Update details (revoked status). - service (
DeviceServiceDep) – Injected device service dependency. - auth (
AdminAuthContextDep) – Authentication context with user info.
Returns:
DeviceSessionResponse– DeviceSessionResponse with the updated session details.
Raises:
403– User does not have access to this session's account.404– Session not found.400– Invalid request or no revoked field provided.
leadr.auth.api.device_session_schemas¶
API schemas for device sessions.
Classes:
- DeviceSessionResponse – Response model for device session.
- DeviceSessionUpdateRequest – Request model for updating device session.
leadr.auth.api.device_session_schemas.DeviceSessionResponse¶
Bases: BaseModel
Response model for device session.
Functions:
- from_domain – Convert domain entity to API response.
Attributes:
- created_at (
datetime) – - device_id (
DeviceID) – - expires_at (
datetime) – - id (
DeviceSessionID) – - ip_address (
str | None) – - refresh_expires_at (
datetime) – - revoked_at (
datetime | None) – - updated_at (
datetime) – - user_agent (
str | None) –
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.created_at¶
created_at: datetime
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.device_id¶
device_id: DeviceID
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.expires_at¶
expires_at: datetime
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.from_domain¶
from_domain(session)
Convert domain entity to API response.
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.id¶
id: DeviceSessionID
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.ip_address¶
ip_address: str | None
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.refresh_expires_at¶
refresh_expires_at: datetime
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.revoked_at¶
revoked_at: datetime | None
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.updated_at¶
updated_at: datetime
# leadr.auth.api.device_session_schemas.DeviceSessionResponse.user_agent¶
user_agent: str | None
leadr.auth.api.device_session_schemas.DeviceSessionUpdateRequest¶
Bases: BaseModel
Request model for updating device session.
Attributes:
# leadr.auth.api.device_session_schemas.DeviceSessionUpdateRequest.revoked¶
revoked: bool | None = None
leadr.auth.bootstrap¶
Superadmin bootstrap functionality.
This module provides functionality to automatically create a superadmin user and associated account on application startup if none exists.
Functions:
- ensure_superadmin_exists – Ensure a superadmin user exists, creating one if necessary.
Attributes:
- logger –
leadr.auth.bootstrap.ensure_superadmin_exists¶
ensure_superadmin_exists(session)
Ensure a superadmin user exists, creating one if necessary.
This function is idempotent and safe to call multiple times. It will:
- Check if any superadmin user already exists
- If not, create:
- A system account (configured via SUPERADMIN_ACCOUNT_NAME/SLUG)
- A superadmin user (configured via SUPERADMIN_EMAIL/DISPLAY_NAME)
- An API key for the superadmin (using SUPERADMIN_API_KEY)
The function commits the transaction if it creates entities.
Parameters:
- session (
AsyncSession) – Database session to use for queries and inserts.
Example
> > > async with get_session() as session: > > > ... await ensure_superadmin_exists(session)leadr.auth.bootstrap.logger¶
logger = logging.getLogger(__name__)
leadr.auth.dependencies¶
Authentication dependencies for FastAPI.
Classes:
- AdminAuthContext – Admin authentication context with guaranteed user and api_key fields.
- AuthContext – Unified authentication context for both admin and client auth.
- AuthContextDependency – Parameterizable authentication dependency using FastAPI class instance pattern.
- ClientAuthContext – Client authentication context with guaranteed device field.
Attributes:
- AdminAuthContextDep –
- AdminAuthContextWithAccountIDDep –
- ClientAuthContextDep –
- ClientAuthContextWithNonceDep –
- logger –
- require_admin_auth –
- require_admin_auth_with_account_id –
- require_client_auth –
- require_client_auth_with_nonce –
leadr.auth.dependencies.AdminAuthContext¶
AdminAuthContext(account_id, user, api_key, device=None)
Bases: AuthContext
Admin authentication context with guaranteed user and api_key fields.
This subclass is returned by admin-only authentication dependencies, providing type-safe access to user and api_key without None checks.
Note: This class does not use @dataclass to avoid conflicts between frozen dataclass fields and property overrides.
Attributes:
- account_id (
AccountID) – The account ID for this request (may differ from API key's account for superadmins). - user (
User) – The authenticated user (guaranteed non-None). - api_key (
APIKey) – The authenticated API key (guaranteed non-None). - device (
None) – Always None for admin auth.
Functions:
- has_access_to_account – Check if the authenticated context has access to a specific account.
leadr.auth.dependencies.AdminAuthContext.account_id¶
account_id: AccountID
Get account ID.
leadr.auth.dependencies.AdminAuthContext.api_key¶
api_key: APIKey
Get API key (guaranteed non-None for admin auth).
leadr.auth.dependencies.AdminAuthContext.auth_type¶
auth_type: Literal['admin', 'client']
Return the authentication type.
Returns:
Literal['admin', 'client']– "admin" if authenticated via API key, "client" if via device token.
leadr.auth.dependencies.AdminAuthContext.device¶
device: None
Get device (always None for admin auth).
leadr.auth.dependencies.AdminAuthContext.has_access_to_account¶
has_access_to_account(account_id)
Check if the authenticated context has access to a specific account.
For admin auth
- Superadmins have access to all accounts - Regular users only have access to their own accountFor client auth
- Devices only have access to their game's accountParameters:
- account_id (
AccountID) – The account ID to check access for.
Returns:
bool– True if context has access to the account, False otherwise.
leadr.auth.dependencies.AdminAuthContext.is_superadmin¶
is_superadmin: bool
Check if the authenticated user has superadmin privileges.
Only applies to admin auth. Client auth never has superadmin privileges.
Returns:
bool– True if user is a superadmin, False otherwise.
leadr.auth.dependencies.AdminAuthContext.user¶
user: User
Get user (guaranteed non-None for admin auth).
leadr.auth.dependencies.AdminAuthContextDep¶
AdminAuthContextDep = Annotated[AdminAuthContext, Depends(require_admin_auth)]
leadr.auth.dependencies.AdminAuthContextWithAccountIDDep¶
AdminAuthContextWithAccountIDDep = Annotated[AdminAuthContext, Depends(require_admin_auth_with_account_id)]
leadr.auth.dependencies.AuthContext¶
AuthContext(account_id, user=None, api_key=None, device=None)
Unified authentication context for both admin and client auth.
This context provides a unified interface for both API key (admin) and device token (client) authentication. It includes helper methods for authorization checks that work transparently across both auth types.
Attributes:
- account_id (
AccountID) – The account ID associated with this auth context. - user (
User | None) – The user entity (present for admin auth only). - api_key (
APIKey | None) – The API key entity (present for admin auth only). - device (
Device | None) – The device entity (present for client auth only).
Functions:
- has_access_to_account – Check if the authenticated context has access to a specific account.
leadr.auth.dependencies.AuthContext.account_id¶
account_id: AccountID
leadr.auth.dependencies.AuthContext.api_key¶
api_key: APIKey | None = None
leadr.auth.dependencies.AuthContext.auth_type¶
auth_type: Literal['admin', 'client']
Return the authentication type.
Returns:
Literal['admin', 'client']– "admin" if authenticated via API key, "client" if via device token.
leadr.auth.dependencies.AuthContext.device¶
device: Device | None = None
leadr.auth.dependencies.AuthContext.has_access_to_account¶
has_access_to_account(account_id)
Check if the authenticated context has access to a specific account.
For admin auth
- Superadmins have access to all accounts - Regular users only have access to their own accountFor client auth
- Devices only have access to their game's accountParameters:
- account_id (
AccountID) – The account ID to check access for.
Returns:
bool– True if context has access to the account, False otherwise.
leadr.auth.dependencies.AuthContext.is_superadmin¶
is_superadmin: bool
Check if the authenticated user has superadmin privileges.
Only applies to admin auth. Client auth never has superadmin privileges.
Returns:
bool– True if user is a superadmin, False otherwise.
leadr.auth.dependencies.AuthContext.user¶
user: User | None = None
leadr.auth.dependencies.AuthContextDependency¶
AuthContextDependency(require_admin=False, require_client=False, require_nonce=False, require_superadmin_account_id=False)
Parameterizable authentication dependency using FastAPI class instance pattern.
This class implements the callable instance pattern to provide flexible authentication requirements. Create instances with different parameters to require different auth types.
Examples:
>>> require_admin_auth = AuthContextDependency(require_admin=True)
>>> require_client_auth = AuthContextDependency(require_client=True)
>>> require_either_auth = AuthContextDependency(require_admin=True, require_client=True)
>>>
>>> @router.get("/admin-only")
>>> async def admin_endpoint(
>>> auth: Annotated[AuthContext, Depends(require_admin_auth)]
>>> ):
>>> return {"account": auth.account_id}
Attributes:
Parameters:
- require_admin (
bool) – If True, admin API key authentication is required. - require_client (
bool) – If True, client device token authentication is required. - require_nonce (
bool) – If True, nonce validation is required for client auth (mutations). - require_superadmin_account_id (
bool) – If True, superadmins must provide account_id query parameter on GET requests. Used for list endpoints.
Raises:
ValueError– If neither require_admin nor require_client is True.
leadr.auth.dependencies.AuthContextDependency.require_admin¶
require_admin = require_admin
leadr.auth.dependencies.AuthContextDependency.require_client¶
require_client = require_client
leadr.auth.dependencies.AuthContextDependency.require_nonce¶
require_nonce = require_nonce
leadr.auth.dependencies.AuthContextDependency.require_superadmin_account_id¶
require_superadmin_account_id = require_superadmin_account_id
leadr.auth.dependencies.ClientAuthContext¶
ClientAuthContext(account_id, device, user=None, api_key=None)
Bases: AuthContext
Client authentication context with guaranteed device field.
This subclass is returned by client-only authentication dependencies, providing type-safe access to device without None checks.
Note: This class does not use @dataclass to avoid conflicts between frozen dataclass fields and property overrides.
Attributes:
- account_id (
AccountID) – The account ID from the device's game. - device (
Device) – The authenticated device (guaranteed non-None). - user (
None) – Always None for client auth. - api_key (
None) – Always None for client auth.
Functions:
- has_access_to_account – Check if the authenticated context has access to a specific account.
leadr.auth.dependencies.ClientAuthContext.account_id¶
account_id: AccountID
Get account ID.
leadr.auth.dependencies.ClientAuthContext.api_key¶
api_key: None
Get API key (always None for client auth).
leadr.auth.dependencies.ClientAuthContext.auth_type¶
auth_type: Literal['admin', 'client']
Return the authentication type.
Returns:
Literal['admin', 'client']– "admin" if authenticated via API key, "client" if via device token.
leadr.auth.dependencies.ClientAuthContext.device¶
device: Device
Get device (guaranteed non-None for client auth).
leadr.auth.dependencies.ClientAuthContext.has_access_to_account¶
has_access_to_account(account_id)
Check if the authenticated context has access to a specific account.
For admin auth
- Superadmins have access to all accounts - Regular users only have access to their own accountFor client auth
- Devices only have access to their game's accountParameters:
- account_id (
AccountID) – The account ID to check access for.
Returns:
bool– True if context has access to the account, False otherwise.
leadr.auth.dependencies.ClientAuthContext.is_superadmin¶
is_superadmin: bool
Check if the authenticated user has superadmin privileges.
Only applies to admin auth. Client auth never has superadmin privileges.
Returns:
bool– True if user is a superadmin, False otherwise.
leadr.auth.dependencies.ClientAuthContext.user¶
user: None
Get user (always None for client auth).
leadr.auth.dependencies.ClientAuthContextDep¶
ClientAuthContextDep = Annotated[ClientAuthContext, Depends(require_client_auth)]
leadr.auth.dependencies.ClientAuthContextWithNonceDep¶
ClientAuthContextWithNonceDep = Annotated[ClientAuthContext, Depends(require_client_auth_with_nonce)]
leadr.auth.dependencies.logger¶
logger = logging.getLogger(__name__)
leadr.auth.dependencies.require_admin_auth¶
require_admin_auth = AuthContextDependency(require_admin=True)
leadr.auth.dependencies.require_admin_auth_with_account_id¶
require_admin_auth_with_account_id = AuthContextDependency(require_admin=True, require_superadmin_account_id=True)
leadr.auth.dependencies.require_client_auth¶
require_client_auth = AuthContextDependency(require_client=True)
leadr.auth.dependencies.require_client_auth_with_nonce¶
require_client_auth_with_nonce = AuthContextDependency(require_client=True, require_nonce=True)
leadr.auth.domain¶
Modules:
- api_key – API Key domain model.
- device – Device domain models for client authentication.
- nonce – Nonce domain entity for replay protection.
leadr.auth.domain.api_key¶
API Key domain model.
Classes:
- APIKey – API Key domain entity.
- APIKeyStatus – API Key status enumeration.
leadr.auth.domain.api_key.APIKey¶
Bases: Entity
API Key domain entity.
Represents an API key used to authenticate requests to the admin API. Each account can have multiple API keys for different purposes. Keys are stored hashed for security and shown only once at creation. Each API key is owned by a specific user within the account.
Functions:
- is_expired – Check if the API key has expired.
- is_valid – Check if the API key is valid for use.
- record_usage – Record that the API key was used.
- restore – Restore a soft-deleted entity.
- revoke – Revoke the API key, preventing further use.
- soft_delete – Mark entity as soft-deleted.
- validate_key_prefix – Validate that key_prefix starts with 'ldr_'.
Attributes:
- account_id (
AccountID) – - created_at (
datetime) – - deleted_at (
datetime | None) – - expires_at (
datetime | None) – - id (
APIKeyID) – - is_deleted (
bool) – Check if entity is soft-deleted. - key_hash (
str) – - key_prefix (
str) – - last_used_at (
datetime | None) – - model_config –
- name (
str) – - status (
APIKeyStatus) – - updated_at (
datetime) – - user_id (
UserID) –
# leadr.auth.domain.api_key.APIKey.account_id¶
account_id: AccountID
# leadr.auth.domain.api_key.APIKey.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
# leadr.auth.domain.api_key.APIKey.deleted_at¶
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
# leadr.auth.domain.api_key.APIKey.expires_at¶
expires_at: datetime | None = None
# leadr.auth.domain.api_key.APIKey.id¶
id: APIKeyID = Field(frozen=True, default_factory=APIKeyID, description='Unique API key identifier')
# leadr.auth.domain.api_key.APIKey.is_deleted¶
is_deleted: bool
Check if entity is soft-deleted.
Returns:
bool– True if the entity has a deleted_at timestamp, False otherwise.
# leadr.auth.domain.api_key.APIKey.is_expired¶
is_expired()
Check if the API key has expired.
Returns:
bool– True if the key has an expiration date and it's in the past.
# leadr.auth.domain.api_key.APIKey.is_valid¶
is_valid()
Check if the API key is valid for use.
A key is valid if it's active and not expired.
Returns:
bool– True if the key can be used for authentication.
# leadr.auth.domain.api_key.APIKey.key_hash¶
key_hash: str
# leadr.auth.domain.api_key.APIKey.key_prefix¶
key_prefix: str
# leadr.auth.domain.api_key.APIKey.last_used_at¶
last_used_at: datetime | None = None
# leadr.auth.domain.api_key.APIKey.model_config¶
model_config = ConfigDict(validate_assignment=True)
# leadr.auth.domain.api_key.APIKey.name¶
name: str
# leadr.auth.domain.api_key.APIKey.record_usage¶
record_usage(used_at)
Record that the API key was used.
Parameters:
- used_at (
datetime) – Timestamp when the key was used.
# leadr.auth.domain.api_key.APIKey.restore¶
restore()
Restore a soft-deleted entity.
Clears the deleted_at timestamp, making the entity active again.
Example
> > > account.soft_delete() > > > account.restore() > > > assert account.is_deleted is False# leadr.auth.domain.api_key.APIKey.revoke¶
revoke()
Revoke the API key, preventing further use.
# leadr.auth.domain.api_key.APIKey.soft_delete¶
soft_delete()
Mark entity as soft-deleted.
Sets the deleted_at timestamp to the current UTC time. Entities that are already deleted are not affected (deleted_at remains at original deletion time).
Example
> > > account = Account(name="Test", slug="test") > > > account.soft_delete() > > > assert account.is_deleted is True# leadr.auth.domain.api_key.APIKey.status¶
status: APIKeyStatus = APIKeyStatus.ACTIVE
# leadr.auth.domain.api_key.APIKey.updated_at¶
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
# leadr.auth.domain.api_key.APIKey.user_id¶
user_id: UserID
# leadr.auth.domain.api_key.APIKey.validate_key_prefix¶
validate_key_prefix(value)
Validate that key_prefix starts with 'ldr_'.
leadr.auth.domain.api_key.APIKeyStatus¶
Bases: Enum
API Key status enumeration.
Attributes:
# leadr.auth.domain.api_key.APIKeyStatus.ACTIVE¶
ACTIVE = 'active'
# leadr.auth.domain.api_key.APIKeyStatus.REVOKED¶
REVOKED = 'revoked'
leadr.auth.domain.device¶
Device domain models for client authentication.
Classes:
- Device – Device domain entity.
- DeviceSession – Device session domain entity.
- DeviceStatus – Device status enumeration.
leadr.auth.domain.device.Device¶
Bases: Entity
Device domain entity.
Represents a game client device (e.g., mobile device, PC, console). Devices are scoped per-game and used for client authentication. Each device is identified by a client-generated SHA256 fingerprint.
Functions:
- activate – Activate the device, allowing authentication.
- ban – Ban the device, preventing further authentication.
- is_active – Check if the device is active.
- restore – Restore a soft-deleted entity.
- soft_delete – Mark entity as soft-deleted.
- suspend – Suspend the device temporarily.
- update_last_seen – Update the last_seen_at timestamp to current time.
- validate_sha256 – Validate that client_fingerprint is a valid SHA256 hash.
Attributes:
- account_id (
AccountID) – - client_fingerprint (
str) – - created_at (
datetime) – - deleted_at (
datetime | None) – - first_seen_at (
datetime) – - game_id (
GameID) – - id (
DeviceID) – - is_deleted (
bool) – Check if entity is soft-deleted. - last_seen_at (
datetime) – - metadata (
dict[str, Any]) – - model_config –
- platform (
str | None) – - status (
DeviceStatus) – - updated_at (
datetime) –
# leadr.auth.domain.device.Device.account_id¶
account_id: AccountID
# leadr.auth.domain.device.Device.activate¶
activate()
Activate the device, allowing authentication.
# leadr.auth.domain.device.Device.ban¶
ban()
Ban the device, preventing further authentication.
# leadr.auth.domain.device.Device.client_fingerprint¶
client_fingerprint: str = Field(description='Client-generated SHA256 device fingerprint (64 hex characters)')
# leadr.auth.domain.device.Device.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
# leadr.auth.domain.device.Device.deleted_at¶
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
# leadr.auth.domain.device.Device.first_seen_at¶
first_seen_at: datetime
# leadr.auth.domain.device.Device.game_id¶
game_id: GameID
# leadr.auth.domain.device.Device.id¶
id: DeviceID = Field(frozen=True, default_factory=DeviceID, description='Unique device identifier')
# leadr.auth.domain.device.Device.is_active¶
is_active()
Check if the device is active.
Returns:
bool– True if the device status is ACTIVE.
# leadr.auth.domain.device.Device.is_deleted¶
is_deleted: bool
Check if entity is soft-deleted.
Returns:
bool– True if the entity has a deleted_at timestamp, False otherwise.
# leadr.auth.domain.device.Device.last_seen_at¶
last_seen_at: datetime
# leadr.auth.domain.device.Device.metadata¶
metadata: dict[str, Any] = {}
# leadr.auth.domain.device.Device.model_config¶
model_config = ConfigDict(validate_assignment=True)
# leadr.auth.domain.device.Device.platform¶
platform: str | None = None
# leadr.auth.domain.device.Device.restore¶
restore()
Restore a soft-deleted entity.
Clears the deleted_at timestamp, making the entity active again.
Example
> > > account.soft_delete() > > > account.restore() > > > assert account.is_deleted is False# leadr.auth.domain.device.Device.soft_delete¶
soft_delete()
Mark entity as soft-deleted.
Sets the deleted_at timestamp to the current UTC time. Entities that are already deleted are not affected (deleted_at remains at original deletion time).
Example
> > > account = Account(name="Test", slug="test") > > > account.soft_delete() > > > assert account.is_deleted is True# leadr.auth.domain.device.Device.status¶
status: DeviceStatus = DeviceStatus.ACTIVE
# leadr.auth.domain.device.Device.suspend¶
suspend()
Suspend the device temporarily.
# leadr.auth.domain.device.Device.update_last_seen¶
update_last_seen()
Update the last_seen_at timestamp to current time.
# leadr.auth.domain.device.Device.updated_at¶
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
# leadr.auth.domain.device.Device.validate_sha256¶
validate_sha256(v)
Validate that client_fingerprint is a valid SHA256 hash.
Parameters:
- v (
str) – The client_fingerprint value to validate.
Returns:
str– The normalized (lowercase) client_fingerprint.
Raises:
ValueError– If the fingerprint is not a valid 64-character hex string.
leadr.auth.domain.device.DeviceSession¶
Bases: Entity
Device session domain entity.
Represents an active authentication session for a device. Sessions have an expiration time and can be revoked manually. Includes both access and refresh tokens with token rotation support.
Functions:
- is_expired – Check if the access token has expired.
- is_refresh_expired – Check if the refresh token has expired.
- is_revoked – Check if the session has been manually revoked.
- is_valid – Check if the session is valid for use.
- restore – Restore a soft-deleted entity.
- revoke – Revoke the session, preventing further use.
- rotate_tokens – Increment token version for token rotation.
- soft_delete – Mark entity as soft-deleted.
Attributes:
- access_token_hash (
str) – - created_at (
datetime) – - deleted_at (
datetime | None) – - device_id (
DeviceID) – - expires_at (
datetime) – - id (
DeviceSessionID) – - ip_address (
str | None) – - is_deleted (
bool) – Check if entity is soft-deleted. - model_config –
- refresh_expires_at (
datetime) – - refresh_token_hash (
str) – - revoked_at (
datetime | None) – - token_version (
int) – - updated_at (
datetime) – - user_agent (
str | None) –
# leadr.auth.domain.device.DeviceSession.access_token_hash¶
access_token_hash: str
# leadr.auth.domain.device.DeviceSession.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
# leadr.auth.domain.device.DeviceSession.deleted_at¶
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
# leadr.auth.domain.device.DeviceSession.device_id¶
device_id: DeviceID
# leadr.auth.domain.device.DeviceSession.expires_at¶
expires_at: datetime
# leadr.auth.domain.device.DeviceSession.id¶
id: DeviceSessionID = Field(frozen=True, default_factory=DeviceSessionID, description='Unique device session identifier')
# leadr.auth.domain.device.DeviceSession.ip_address¶
ip_address: str | None = None
# leadr.auth.domain.device.DeviceSession.is_deleted¶
is_deleted: bool
Check if entity is soft-deleted.
Returns:
bool– True if the entity has a deleted_at timestamp, False otherwise.
# leadr.auth.domain.device.DeviceSession.is_expired¶
is_expired()
Check if the access token has expired.
Returns:
bool– True if the current time is past the expiration time.
# leadr.auth.domain.device.DeviceSession.is_refresh_expired¶
is_refresh_expired()
Check if the refresh token has expired.
Returns:
bool– True if the current time is past the refresh expiration time.
# leadr.auth.domain.device.DeviceSession.is_revoked¶
is_revoked()
Check if the session has been manually revoked.
Returns:
bool– True if revoked_at is set.
# leadr.auth.domain.device.DeviceSession.is_valid¶
is_valid()
Check if the session is valid for use.
A session is valid if it's not expired and not revoked.
Returns:
bool– True if the session can be used for authentication.
# leadr.auth.domain.device.DeviceSession.model_config¶
model_config = ConfigDict(validate_assignment=True)
# leadr.auth.domain.device.DeviceSession.refresh_expires_at¶
refresh_expires_at: datetime
# leadr.auth.domain.device.DeviceSession.refresh_token_hash¶
refresh_token_hash: str
# leadr.auth.domain.device.DeviceSession.restore¶
restore()
Restore a soft-deleted entity.
Clears the deleted_at timestamp, making the entity active again.
Example
> > > account.soft_delete() > > > account.restore() > > > assert account.is_deleted is False# leadr.auth.domain.device.DeviceSession.revoke¶
revoke()
Revoke the session, preventing further use.
# leadr.auth.domain.device.DeviceSession.revoked_at¶
revoked_at: datetime | None = None
# leadr.auth.domain.device.DeviceSession.rotate_tokens¶
rotate_tokens()
Increment token version for token rotation.
Called when refreshing tokens to invalidate old refresh tokens.
# leadr.auth.domain.device.DeviceSession.soft_delete¶
soft_delete()
Mark entity as soft-deleted.
Sets the deleted_at timestamp to the current UTC time. Entities that are already deleted are not affected (deleted_at remains at original deletion time).
Example
> > > account = Account(name="Test", slug="test") > > > account.soft_delete() > > > assert account.is_deleted is True# leadr.auth.domain.device.DeviceSession.token_version¶
token_version: int = 1
# leadr.auth.domain.device.DeviceSession.updated_at¶
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
# leadr.auth.domain.device.DeviceSession.user_agent¶
user_agent: str | None = None
leadr.auth.domain.device.DeviceStatus¶
Bases: Enum
Device status enumeration.
Attributes:
# leadr.auth.domain.device.DeviceStatus.ACTIVE¶
ACTIVE = 'active'
# leadr.auth.domain.device.DeviceStatus.BANNED¶
BANNED = 'banned'
# leadr.auth.domain.device.DeviceStatus.SUSPENDED¶
SUSPENDED = 'suspended'
leadr.auth.domain.nonce¶
Nonce domain entity for replay protection.
Classes:
- Nonce – Request nonce for replay protection.
- NonceStatus – Nonce status enumeration.
leadr.auth.domain.nonce.Nonce¶
Bases: Entity
Request nonce for replay protection.
Nonces are single-use tokens that clients must obtain before making mutating requests (POST, PATCH, DELETE). Each nonce has a short TTL (typically 60 seconds) and can only be used once.
This prevents replay attacks by ensuring that each mutating request is fresh and authorized by the server.
Functions:
- is_expired – Check if nonce has expired.
- is_used – Check if nonce has been used.
- is_valid – Check if nonce is valid (not used and not expired).
- mark_expired – Mark nonce as expired.
- mark_used – Mark nonce as used.
- restore – Restore a soft-deleted entity.
- soft_delete – Mark entity as soft-deleted.
Attributes:
- created_at (
datetime) – - deleted_at (
datetime | None) – - device_id (
DeviceID) – - expires_at (
datetime) – - id (
NonceID) – - is_deleted (
bool) – Check if entity is soft-deleted. - model_config –
- nonce_value (
str) – - status (
NonceStatus) – - updated_at (
datetime) – - used_at (
datetime | None) –
# leadr.auth.domain.nonce.Nonce.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
# leadr.auth.domain.nonce.Nonce.deleted_at¶
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
# leadr.auth.domain.nonce.Nonce.device_id¶
device_id: DeviceID = Field(description='Device that owns this nonce')
# leadr.auth.domain.nonce.Nonce.expires_at¶
expires_at: datetime = Field(description='Nonce expiration timestamp')
# leadr.auth.domain.nonce.Nonce.id¶
id: NonceID = Field(frozen=True, default_factory=NonceID, description='Unique nonce identifier')
# leadr.auth.domain.nonce.Nonce.is_deleted¶
is_deleted: bool
Check if entity is soft-deleted.
Returns:
bool– True if the entity has a deleted_at timestamp, False otherwise.
# leadr.auth.domain.nonce.Nonce.is_expired¶
is_expired()
Check if nonce has expired.
Returns:
bool– True if current time is at or past expires_at
# leadr.auth.domain.nonce.Nonce.is_used¶
is_used()
Check if nonce has been used.
Returns:
bool– True if status is USED
# leadr.auth.domain.nonce.Nonce.is_valid¶
is_valid()
Check if nonce is valid (not used and not expired).
Returns:
bool– True if nonce is pending and not expired
# leadr.auth.domain.nonce.Nonce.mark_expired¶
mark_expired()
Mark nonce as expired.
Only marks nonce as expired if it's currently pending. Does not change status if already used or expired.
# leadr.auth.domain.nonce.Nonce.mark_used¶
mark_used()
Mark nonce as used.
Sets status to USED and records used_at timestamp.
Raises:
ValueError– If nonce is not valid (already used or expired)
# leadr.auth.domain.nonce.Nonce.model_config¶
model_config = ConfigDict(validate_assignment=True)
# leadr.auth.domain.nonce.Nonce.nonce_value¶
nonce_value: str = Field(description='Unique nonce value (UUID string)')
# leadr.auth.domain.nonce.Nonce.restore¶
restore()
Restore a soft-deleted entity.
Clears the deleted_at timestamp, making the entity active again.
Example
> > > account.soft_delete() > > > account.restore() > > > assert account.is_deleted is False# leadr.auth.domain.nonce.Nonce.soft_delete¶
soft_delete()
Mark entity as soft-deleted.
Sets the deleted_at timestamp to the current UTC time. Entities that are already deleted are not affected (deleted_at remains at original deletion time).
Example
> > > account = Account(name="Test", slug="test") > > > account.soft_delete() > > > assert account.is_deleted is True# leadr.auth.domain.nonce.Nonce.status¶
status: NonceStatus = Field(default=(NonceStatus.PENDING), description='Nonce status')
# leadr.auth.domain.nonce.Nonce.updated_at¶
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
# leadr.auth.domain.nonce.Nonce.used_at¶
used_at: datetime | None = Field(default=None, description='When nonce was consumed')
leadr.auth.domain.nonce.NonceStatus¶
Nonce status enumeration.
Attributes:
# leadr.auth.domain.nonce.NonceStatus.EXPIRED¶
EXPIRED = 'expired'
# leadr.auth.domain.nonce.NonceStatus.PENDING¶
PENDING = 'pending'
# leadr.auth.domain.nonce.NonceStatus.USED¶
USED = 'used'
leadr.auth.services¶
Modules:
- api_key_crypto – Cryptographic operations for API keys.
- api_key_service – API Key service for managing API key operations.
- dependencies – Auth service dependency injection factories.
- device_service – Device authentication service.
- device_token_crypto – Cryptographic operations for device access and refresh tokens.
- nonce_service – Nonce service for managing request nonces.
- nonce_tasks – Background tasks for nonce cleanup.
- repositories – API Key, Device, and Nonce repository services.
leadr.auth.services.api_key_crypto¶
Cryptographic operations for API keys.
Functions:
- generate_api_key – Generate a secure random API key with 'ldr_' prefix.
- hash_api_key – Hash an API key for secure storage using HMAC-SHA256.
- verify_api_key – Verify an API key against its stored hash.
Attributes:
leadr.auth.services.api_key_crypto.API_KEY_BYTES¶
API_KEY_BYTES = 32
leadr.auth.services.api_key_crypto.API_KEY_PREFIX¶
API_KEY_PREFIX = 'ldr_'
leadr.auth.services.api_key_crypto.generate_api_key¶
generate_api_key()
Generate a secure random API key with 'ldr_' prefix.
The key is generated using cryptographically secure random bytes and encoded using URL-safe base64 encoding for compatibility.
Returns:
str– A secure API key string starting with 'ldr_' followed bystr– URL-safe random characters (alphanumeric, hyphen, underscore).
Example
> > > key = generate_api_key() > > > key.startswith('ldr\_') > > > True > > > len(key) > 36 > > > Trueleadr.auth.services.api_key_crypto.hash_api_key¶
hash_api_key(key, secret)
Hash an API key for secure storage using HMAC-SHA256.
Uses HMAC-SHA256 with a server-side secret (pepper) to create a one-way hash of the API key. This provides defense in depth: database compromise alone isn't enough to validate keys.
Parameters:
Returns:
str– A hexadecimal string representation of the HMAC-SHA256 hash.
Example
> > > secret = "my-secret" > > > hash1 = hash_api_key('ldr_test123', secret) > > > hash2 = hash_api_key('ldr_test123', secret) > > > hash1 == hash2 > > > True > > > len(hash1) > > > 64leadr.auth.services.api_key_crypto.verify_api_key¶
verify_api_key(key, key_hash, secret)
Verify an API key against its stored hash.
Uses timing-safe comparison to prevent timing attacks.
Parameters:
- key (
str) – The API key to verify. - key_hash (
str) – The stored hash to compare against. - secret (
str) – Server-side secret for HMAC verification.
Returns:
bool– True if the key matches the hash, False otherwise.
Example
> > > secret = "my-secret" > > > key = 'ldr_test123' > > > hashed = hash_api_key(key, secret) > > > verify_api_key(key, hashed, secret) > > > True > > > verify_api_key('ldr_wrong', hashed, secret) > > > Falseleadr.auth.services.api_key_service¶
API Key service for managing API key operations.
Classes:
- APIKeyService – Service for managing API key lifecycle and operations.
leadr.auth.services.api_key_service.APIKeyService¶
Bases: BaseService[APIKey, APIKeyRepository]
Service for managing API key lifecycle and operations.
This service orchestrates API key creation, validation, and management by coordinating between the domain models, cryptographic functions, and repository layer.
Functions:
- count_active_api_keys – Count active API keys for an account.
- create_api_key – Create a new API key for a user within an account.
- create_api_key_with_value – Create a new API key with a specific key value (for bootstrap/testing).
- delete – Soft-delete an entity.
- get_api_key – Get an API key by its ID.
- get_by_id – Get an entity by its ID.
- get_by_id_or_raise – Get an entity by its ID or raise EntityNotFoundError.
- list_account_api_keys – List all API keys for an account.
- list_all – List all non-deleted entities.
- list_api_keys – List API keys for an account with optional filters and pagination.
- record_usage – Record that an API key was used at a specific time.
- revoke_api_key – Revoke an API key, preventing further use.
- soft_delete – Soft-delete an entity and return it before deletion.
- update_api_key_status – Update the status of an API key.
- validate_api_key – Validate an API key and return the domain entity if valid.
Attributes:
# leadr.auth.services.api_key_service.APIKeyService.count_active_api_keys¶
count_active_api_keys(account_id)
Count active API keys for an account.
This is useful for enforcing limits on the number of active keys per account based on their plan or tier.
Parameters:
- account_id (
AccountID) – The account ID to count keys for.
Returns:
int– Number of active (non-revoked) API keys.
# leadr.auth.services.api_key_service.APIKeyService.create_api_key¶
create_api_key(account_id, user_id, name, expires_at=None)
Create a new API key for a user within an account.
Generates a secure random key, hashes it for storage, and persists it to the database. The plain key is returned only once for the caller to provide to the user.
Parameters:
- account_id (
AccountID) – The account ID the key belongs to. - user_id (
UserID) – The user ID the key belongs to. - name (
str) – A descriptive name for the key. - expires_at (
datetime | None) – Optional expiration timestamp for the key.
Returns:
APIKey– A tuple of (APIKey domain entity, plain key string).str– The plain key should be shown to the user once and not stored.
Example
> > > api_key, plain_key = await service.create_api_key( > > > ... account_id=account_id, > > > ... user_id=user_id, > > > ... name="Production API Key", > > > ... expires_at=datetime.now(UTC) + timedelta(days=90) > > > ... ) > > > print(f"Your API key: {plain_key}") > > > Your API key: ldr_abc123...# leadr.auth.services.api_key_service.APIKeyService.create_api_key_with_value¶
create_api_key_with_value(account_id, user_id, name, key_value, expires_at=None)
Create a new API key with a specific key value (for bootstrap/testing).
This method is used for creating API keys with predetermined values, such as during superadmin bootstrap. Unlike create_api_key, this does not generate a random key and only returns the APIKey entity.
Parameters:
- account_id (
AccountID) – The account ID the key belongs to. - user_id (
UserID) – The user ID the key belongs to. - name (
str) – A descriptive name for the key. - key_value (
str) – The specific API key value to use. - expires_at (
datetime | None) – Optional expiration timestamp for the key.
Returns:
APIKey– The created APIKey domain entity.
Example
> > > api_key = await service.create_api_key_with_value( > > > ... account_id=account_id, > > > ... user_id=user_id, > > > ... name="Superadmin Key", > > > ... key_value="ldr_fixed_key_for_bootstrap", > > > ... )# leadr.auth.services.api_key_service.APIKeyService.delete¶
delete(entity_id)
Soft-delete an entity.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to delete
Raises:
EntityNotFoundError– If the entity doesn't exist
# leadr.auth.services.api_key_service.APIKeyService.get_api_key¶
get_api_key(key_id)
Get an API key by its ID.
Parameters:
- key_id (
APIKeyID) – The ID of the API key to retrieve.
Returns:
APIKey | None– The APIKey domain entity if found, None otherwise.
# leadr.auth.services.api_key_service.APIKeyService.get_by_id¶
get_by_id(entity_id)
Get an entity by its ID.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to retrieve
Returns:
DomainEntityT | None– The domain entity if found, None otherwise
# leadr.auth.services.api_key_service.APIKeyService.get_by_id_or_raise¶
get_by_id_or_raise(entity_id)
Get an entity by its ID or raise EntityNotFoundError.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to retrieve
Returns:
DomainEntityT– The domain entity
Raises:
EntityNotFoundError– If the entity is not found (converted to HTTP 404 by global handler)
# leadr.auth.services.api_key_service.APIKeyService.list_account_api_keys¶
list_account_api_keys(account_id, active_only=False)
List all API keys for an account.
Parameters:
- account_id (
AccountID) – The account ID to list keys for. - active_only (
bool) – If True, only return active (non-revoked) keys.
Returns:
# leadr.auth.services.api_key_service.APIKeyService.list_all¶
list_all()
List all non-deleted entities.
Returns:
list[DomainEntityT]– List of domain entities
# leadr.auth.services.api_key_service.APIKeyService.list_api_keys¶
list_api_keys(account_id, status=None, pagination=None)
List API keys for an account with optional filters and pagination.
Parameters:
- account_id (
AccountID | None) – Account ID to filter by. If None, returns all API keys (superadmin use case). - status (
str | None) – Optional status string to filter by. - pagination (
PaginationParams | None) – Optional pagination parameters.
Returns:
list[APIKey] | PaginatedResult[APIKey]– List of APIKey entities if no pagination, PaginatedResult if pagination provided.
# leadr.auth.services.api_key_service.APIKeyService.record_usage¶
record_usage(key_id, used_at)
Record that an API key was used at a specific time.
This is typically called automatically during validation, but can also be called explicitly if needed.
Parameters:
- key_id (
APIKeyID) – The ID of the API key that was used. - used_at (
datetime) – The timestamp when the key was used.
Returns:
APIKey– The updated APIKey domain entity.
Raises:
EntityNotFoundError– If the key doesn't exist.
# leadr.auth.services.api_key_service.APIKeyService.repository¶
repository = self._create_repository(session)
# leadr.auth.services.api_key_service.APIKeyService.revoke_api_key¶
revoke_api_key(key_id)
Revoke an API key, preventing further use.
Parameters:
- key_id (
APIKeyID) – The ID of the API key to revoke.
Returns:
APIKey– The updated APIKey domain entity with REVOKED status.
Raises:
EntityNotFoundError– If the key doesn't exist.
# leadr.auth.services.api_key_service.APIKeyService.soft_delete¶
soft_delete(entity_id)
Soft-delete an entity and return it before deletion.
Useful for endpoints that need to return the deleted entity in the response.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to delete
Returns:
DomainEntityT– The entity before it was deleted
Raises:
EntityNotFoundError– If the entity doesn't exist
# leadr.auth.services.api_key_service.APIKeyService.update_api_key_status¶
update_api_key_status(key_id, status)
Update the status of an API key.
Parameters:
- key_id (
APIKeyID) – The ID of the API key to update. - status (
str) – The new status value (active or revoked).
Returns:
APIKey– The updated APIKey domain entity.
Raises:
EntityNotFoundError– If the key doesn't exist.ValueError– If the status is invalid.
# leadr.auth.services.api_key_service.APIKeyService.validate_api_key¶
validate_api_key(plain_key)
Validate an API key and return the domain entity if valid.
Performs the following checks:
- Extracts prefix and looks up key in database
- Verifies the hash matches
- Checks if key is active (not revoked)
- Checks if key is not expired
- Records usage timestamp if valid
Parameters:
- plain_key (
str) – The plain API key string to validate.
Returns:
APIKey | None– The APIKey domain entity if valid, None otherwise.
Example
> > > api_key = await service.validate_api_key("ldr_abc123...") > > > if api_key: > > > ... print(f"Valid key for account {api_key.account_id}") > > > ... else: > > > ... print("Invalid or expired key")leadr.auth.services.dependencies¶
Auth service dependency injection factories.
Functions:
- get_api_key_service – Get APIKeyService dependency.
- get_device_service – Get DeviceService dependency.
- get_nonce_service – Get NonceService dependency.
Attributes:
leadr.auth.services.dependencies.APIKeyServiceDep¶
APIKeyServiceDep = Annotated[APIKeyService, Depends(get_api_key_service)]
leadr.auth.services.dependencies.DeviceServiceDep¶
DeviceServiceDep = Annotated[DeviceService, Depends(get_device_service)]
leadr.auth.services.dependencies.NonceServiceDep¶
NonceServiceDep = Annotated[NonceService, Depends(get_nonce_service)]
leadr.auth.services.dependencies.get_api_key_service¶
get_api_key_service(db)
Get APIKeyService dependency.
Parameters:
- db (
DatabaseSession) – Database session injected via dependency injection
Returns:
APIKeyService– APIKeyService instance configured with the database session
leadr.auth.services.dependencies.get_device_service¶
get_device_service(db)
Get DeviceService dependency.
Parameters:
- db (
DatabaseSession) – Database session injected via dependency injection
Returns:
DeviceService– DeviceService instance configured with the database session
leadr.auth.services.dependencies.get_nonce_service¶
get_nonce_service(db)
Get NonceService dependency.
Parameters:
- db (
DatabaseSession) – Database session injected via dependency injection
Returns:
NonceService– NonceService instance configured with the database session
leadr.auth.services.device_service¶
Device authentication service.
Classes:
- DeviceService – Service for device authentication and session management.
leadr.auth.services.device_service.DeviceService¶
DeviceService(session)
Bases: BaseService[Device, DeviceRepository]
Service for device authentication and session management.
Functions:
- activate_device – Activate a device, allowing authentication.
- ban_device – Ban a device, preventing further authentication.
- delete – Soft-delete an entity.
- get_by_id – Get an entity by its ID.
- get_by_id_or_raise – Get an entity by its ID or raise EntityNotFoundError.
- get_device – Get a device by its ID.
- get_session – Get a device session by its ID.
- get_session_or_raise – Get a device session by its ID or raise EntityNotFoundError.
- list_all – List all non-deleted entities.
- list_devices – List devices for an account with optional filters and pagination.
- list_sessions – List device sessions for an account with optional filters and pagination.
- refresh_access_token – Refresh access token using a valid refresh token.
- revoke_session – Revoke a device session.
- soft_delete – Soft-delete an entity and return it before deletion.
- start_session – Start a new device session.
- suspend_device – Suspend a device temporarily.
- validate_device_token – Validate access token and return associated device.
Attributes:
- repository –
- session –
- session_repo –
Parameters:
- session (
AsyncSession) – SQLAlchemy async session
# leadr.auth.services.device_service.DeviceService.activate_device¶
activate_device(device_id)
Activate a device, allowing authentication.
Parameters:
- device_id (
DeviceID) – The ID of the device to activate
Returns:
Device– The updated device
Raises:
EntityNotFoundError– If the device doesn't exist
Example
> > > device = await service.activate_device(device_id)# leadr.auth.services.device_service.DeviceService.ban_device¶
ban_device(device_id)
Ban a device, preventing further authentication.
Parameters:
- device_id (
DeviceID) – The ID of the device to ban
Returns:
Device– The updated device
Raises:
EntityNotFoundError– If the device doesn't exist
Example
> > > device = await service.ban_device(device_id)# leadr.auth.services.device_service.DeviceService.delete¶
delete(entity_id)
Soft-delete an entity.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to delete
Raises:
EntityNotFoundError– If the entity doesn't exist
# leadr.auth.services.device_service.DeviceService.get_by_id¶
get_by_id(entity_id)
Get an entity by its ID.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to retrieve
Returns:
DomainEntityT | None– The domain entity if found, None otherwise
# leadr.auth.services.device_service.DeviceService.get_by_id_or_raise¶
get_by_id_or_raise(entity_id)
Get an entity by its ID or raise EntityNotFoundError.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to retrieve
Returns:
DomainEntityT– The domain entity
Raises:
EntityNotFoundError– If the entity is not found (converted to HTTP 404 by global handler)
# leadr.auth.services.device_service.DeviceService.get_device¶
get_device(device_id)
Get a device by its ID.
Parameters:
- device_id (
UUID) – The ID of the device to retrieve
Returns:
Device | None– The device if found, None otherwise
Example
> > > device = await service.get_device(device_id)# leadr.auth.services.device_service.DeviceService.get_session¶
get_session(session_id)
Get a device session by its ID.
Parameters:
- session_id (
UUID) – The ID of the session to retrieve
Returns:
DeviceSession | None– The session if found, None otherwise
Example
> > > session = await service.get_session(session_id)# leadr.auth.services.device_service.DeviceService.get_session_or_raise¶
get_session_or_raise(session_id)
Get a device session by its ID or raise EntityNotFoundError.
Parameters:
- session_id (
DeviceSessionID) – The ID of the session to retrieve
Returns:
DeviceSession– The session
Raises:
EntityNotFoundError– If the session doesn't exist
Example
> > > session = await service.get_session_or_raise(session_id)# leadr.auth.services.device_service.DeviceService.list_all¶
list_all()
List all non-deleted entities.
Returns:
list[DomainEntityT]– List of domain entities
# leadr.auth.services.device_service.DeviceService.list_devices¶
list_devices(account_id, game_id=None, status=None, pagination=None)
List devices for an account with optional filters and pagination.
Parameters:
- account_id (
AccountID | None) – Account ID to filter by. If None, returns all devices (superadmin use case). - game_id (
GameID | None) – Optional game ID to filter by - status (
str | None) – Optional status to filter by (active, banned, suspended) - pagination (
PaginationParams | None) – Optional pagination parameters
Returns:
list[Device] | PaginatedResult[Device]– List of Device entities if no pagination, PaginatedResult if pagination provided
Example
> > > devices = await service.list_devices( > > > ... account_id=account.id, > > > ... status="active", > > > ... )# leadr.auth.services.device_service.DeviceService.list_sessions¶
list_sessions(account_id, device_id=None, pagination=None)
List device sessions for an account with optional filters and pagination.
Parameters:
- account_id (
AccountID | None) – Account ID to filter by. If None, returns all sessions (superadmin use case). - device_id (
DeviceID | None) – Optional device ID to filter by - pagination (
PaginationParams | None) – Optional pagination parameters
Returns:
list[DeviceSession] | PaginatedResult[DeviceSession]– List of DeviceSession entities if no pagination, PaginatedResult if pagination provided
Example
> > > sessions = await service.list_sessions( > > > ... account_id=account.id, > > > ... device_id=device.id, > > > ... )# leadr.auth.services.device_service.DeviceService.refresh_access_token¶
refresh_access_token(refresh_token)
Refresh access token using a valid refresh token.
Validates the refresh token, checks token version for replay attack detection, generates new access and refresh tokens with incremented version, and updates the session.
Parameters:
- refresh_token (
str) – JWT refresh token
Returns:
tuple[str, str, int] | None– tuple[str, str, int]: (access_token_plain, refresh_token_plain, expires_in_seconds)tuple[str, str, int] | None– or None if refresh token is invalid
Token Rotation Security
- The token_version in the JWT must match the session's token_version - When tokens are refreshed, the version is incremented - Old refresh tokens with lower versions are rejected (prevents replay attacks)# leadr.auth.services.device_service.DeviceService.repository¶
repository = self._create_repository(session)
# leadr.auth.services.device_service.DeviceService.revoke_session¶
revoke_session(session_id)
Revoke a device session.
Parameters:
- session_id (
DeviceSessionID) – The ID of the session to revoke
Returns:
DeviceSession– The updated session
Raises:
EntityNotFoundError– If the session doesn't exist
Example
> > > session = await service.revoke_session(session_id)# leadr.auth.services.device_service.DeviceService.session¶
session = session
# leadr.auth.services.device_service.DeviceService.session_repo¶
session_repo = DeviceSessionRepository(session)
# leadr.auth.services.device_service.DeviceService.soft_delete¶
soft_delete(entity_id)
Soft-delete an entity and return it before deletion.
Useful for endpoints that need to return the deleted entity in the response.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to delete
Returns:
DomainEntityT– The entity before it was deleted
Raises:
EntityNotFoundError– If the entity doesn't exist
# leadr.auth.services.device_service.DeviceService.start_session¶
start_session(game_id, client_fingerprint, platform=None, ip_address=None, user_agent=None, metadata=None)
Start a new device session.
Creates or updates device, generates JWT access and refresh tokens, and creates session record. This is idempotent - calling multiple times updates last_seen_at.
Parameters:
- game_id (
GameID) – Game UUID - client_fingerprint (
str) – Client-generated SHA256 device fingerprint - platform (
str | None) – Device platform (ios, android, etc.) - ip_address (
str | None) – Client IP address - user_agent (
str | None) – Client user agent string - metadata (
dict[str, Any] | None) – Additional device metadata
Returns:
tuple[Device, str, str, int]– tuple[Device, str, str, int]: (device, access_token_plain, refresh_token_plain, expires_in_seconds)
Raises:
EntityNotFoundError– If game doesn't exist
# leadr.auth.services.device_service.DeviceService.suspend_device¶
suspend_device(device_id)
Suspend a device temporarily.
Parameters:
- device_id (
DeviceID) – The ID of the device to suspend
Returns:
Device– The updated device
Raises:
EntityNotFoundError– If the device doesn't exist
Example
> > > device = await service.suspend_device(device_id)# leadr.auth.services.device_service.DeviceService.validate_device_token¶
validate_device_token(token)
Validate access token and return associated device.
Validates JWT signature and expiration, checks session validity, and ensures device is active.
Parameters:
- token (
str) – JWT access token
Returns:
Device | None– Device if token is valid and device is active, None otherwise
leadr.auth.services.device_token_crypto¶
Cryptographic operations for device access and refresh tokens.
Functions:
- generate_access_token – Generate JWT access token for device authentication.
- generate_refresh_token – Generate JWT refresh token for device authentication.
- hash_token – Hash token for secure storage using HMAC-SHA256.
- validate_access_token – Validate and decode JWT access token.
- validate_refresh_token – Validate and decode JWT refresh token.
leadr.auth.services.device_token_crypto.generate_access_token¶
generate_access_token(client_fingerprint, game_id, account_id, expires_delta, secret)
Generate JWT access token for device authentication.
Creates a JWT with device, game, and account claims, signs it with the secret, and returns both the plain token and its SHA-256 hash for storage.
Parameters:
- device_id – Client-generated SHA256 device fingerprint
- game_id (
GameID) – Game UUID - account_id (
AccountID) – Account UUID (for multi-tenant isolation) - expires_delta (
timedelta) – Time until token expires - secret (
str) – Server-side secret for JWT signing
Returns:
tuple[str, str]– tuple[str, str]: (token_plain, token_hash)- token_plain: JWT access token to return to client
- token_hash: SHA-256 hash for secure storage
Example
> > > device_id = "device-123" > > > game_id = UUID("...") > > > account_id = UUID("...") > > > token, token_hash = generate_access_token( > > > ... device_id, game_id, account_id, timedelta(hours=1), "secret" > > > ... ) > > > token.count(".") > > > 2leadr.auth.services.device_token_crypto.generate_refresh_token¶
generate_refresh_token(client_fingerprint, game_id, account_id, token_version, expires_delta, secret)
Generate JWT refresh token for device authentication.
Creates a JWT with device, game, account, and version claims, signs it with the secret, and returns both the plain token and its SHA-256 hash for storage.
The token_version claim enables token rotation: when a refresh token is used, the version is incremented and old tokens with lower versions are invalidated.
Parameters:
- device_id – Client-generated SHA256 device fingerprint
- game_id (
GameID) – Game UUID - account_id (
AccountID) – Account UUID (for multi-tenant isolation) - token_version (
int) – Current token version for rotation tracking - expires_delta (
timedelta) – Time until token expires (typically 30 days) - secret (
str) – Server-side secret for JWT signing
Returns:
tuple[str, str]– tuple[str, str]: (token_plain, token_hash)- token_plain: JWT refresh token to return to client
- token_hash: SHA-256 hash for secure storage
Example
> > > device_id = "device-123" > > > game_id = UUID("...") > > > account_id = UUID("...") > > > token, token_hash = generate_refresh_token( > > > ... device_id, game_id, account_id, 1, timedelta(days=30), "secret" > > > ... ) > > > token.count(".") > > > 2leadr.auth.services.device_token_crypto.hash_token¶
hash_token(token, secret)
Hash token for secure storage using HMAC-SHA256.
Uses HMAC-SHA256 with a server-side secret (pepper) to create a one-way hash of the token. This provides defense in depth: database compromise alone isn't enough to use tokens.
Parameters:
Returns:
str– A hexadecimal string representation of the HMAC-SHA256 hash
Example
> > > secret = "my-secret" > > > hash1 = hash_token("token123", secret) > > > hash2 = hash_token("token123", secret) > > > hash1 == hash2 > > > True > > > len(hash1) > > > 64leadr.auth.services.device_token_crypto.validate_access_token¶
validate_access_token(token, secret)
Validate and decode JWT access token.
Verifies the token signature and expiration. Returns decoded claims if valid.
Parameters:
Returns:
dict[str, Any] | None– dict with claims (sub, game_id, account_id, exp, iat, jti) or None if invalid
Example
> > > token = "eyJ..." > > > claims = validate_access_token(token, "secret") > > > claims["sub"] if claims else None > > > 'device-123'leadr.auth.services.device_token_crypto.validate_refresh_token¶
validate_refresh_token(token, secret)
Validate and decode JWT refresh token.
Verifies the token signature and expiration. Returns decoded claims if valid.
Parameters:
Returns:
dict[str, Any] | None– dict with claims (sub, game_id, account_id, token_version, exp, iat, jti) or None if invalid
Example
> > > token = "eyJ..." > > > claims = validate_refresh_token(token, "secret") > > > claims["token_version"] if claims else None > > > 1leadr.auth.services.nonce_service¶
Nonce service for managing request nonces.
Classes:
- NonceService – Service for managing request nonces.
leadr.auth.services.nonce_service.NonceService¶
Bases: BaseService[Nonce, NonceRepository]
Service for managing request nonces.
Nonces are single-use tokens that clients must obtain before making mutating requests. This prevents replay attacks by ensuring each request is fresh and authorized by the server.
Functions:
- cleanup_expired_nonces – Clean up expired nonces older than specified hours.
- delete – Soft-delete an entity.
- generate_nonce – Generate a fresh nonce for a device.
- get_by_id – Get an entity by its ID.
- get_by_id_or_raise – Get an entity by its ID or raise EntityNotFoundError.
- list_all – List all non-deleted entities.
- soft_delete – Soft-delete an entity and return it before deletion.
- validate_and_consume_nonce – Validate nonce and mark as used (atomic operation).
Attributes:
# leadr.auth.services.nonce_service.NonceService.cleanup_expired_nonces¶
cleanup_expired_nonces(older_than_hours=24)
Clean up expired nonces older than specified hours.
Only deletes nonces with PENDING status. Used nonces are kept for audit/debugging purposes.
Parameters:
- older_than_hours (
int) – Delete nonces expired before this many hours ago (default 24)
Returns:
int– Number of nonces deleted
Example
> > > # In background task or cron job > > > > > > deleted = await service.cleanup_expired_nonces(older_than_hours=24) > > > logger.info(f"Cleaned up {deleted} expired nonces")# leadr.auth.services.nonce_service.NonceService.delete¶
delete(entity_id)
Soft-delete an entity.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to delete
Raises:
EntityNotFoundError– If the entity doesn't exist
# leadr.auth.services.nonce_service.NonceService.generate_nonce¶
generate_nonce(device_id, ttl_seconds=60)
Generate a fresh nonce for a device.
Parameters:
- device_id (
DeviceID) – Device ID to associate nonce with - ttl_seconds (
int) – Time-to-live in seconds (default 60)
Returns:
Example
> > > nonce_value, expires_at = await service.generate_nonce(device_id) > > > > > > # Client includes nonce_value in leadr-client-nonce header# leadr.auth.services.nonce_service.NonceService.get_by_id¶
get_by_id(entity_id)
Get an entity by its ID.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to retrieve
Returns:
DomainEntityT | None– The domain entity if found, None otherwise
# leadr.auth.services.nonce_service.NonceService.get_by_id_or_raise¶
get_by_id_or_raise(entity_id)
Get an entity by its ID or raise EntityNotFoundError.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to retrieve
Returns:
DomainEntityT– The domain entity
Raises:
EntityNotFoundError– If the entity is not found (converted to HTTP 404 by global handler)
# leadr.auth.services.nonce_service.NonceService.list_all¶
list_all()
List all non-deleted entities.
Returns:
list[DomainEntityT]– List of domain entities
# leadr.auth.services.nonce_service.NonceService.repository¶
repository = self._create_repository(session)
# leadr.auth.services.nonce_service.NonceService.soft_delete¶
soft_delete(entity_id)
Soft-delete an entity and return it before deletion.
Useful for endpoints that need to return the deleted entity in the response.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to delete
Returns:
DomainEntityT– The entity before it was deleted
Raises:
EntityNotFoundError– If the entity doesn't exist
# leadr.auth.services.nonce_service.NonceService.validate_and_consume_nonce¶
validate_and_consume_nonce(nonce_value, device_id)
Validate nonce and mark as used (atomic operation).
Parameters:
- nonce_value (
str) – The nonce value to validate - device_id (
DeviceID) – Expected device ID (must match nonce owner)
Returns:
bool– True if nonce was valid and consumed
Raises:
ValueError– If nonce is invalid (expired, already used, wrong device, or not found)
Example
> > > try: > > > ... await service.validate_and_consume_nonce(nonce_value, device.id) > > > ... except ValueError as e: > > > ... # Handle invalid nonce (return 412 error to client) > > > ... raise HTTPException(status_code=412, detail=str(e))leadr.auth.services.nonce_tasks¶
Background tasks for nonce cleanup.
Contains tasks for:
- Cleaning up expired nonces to prevent database bloat
Functions:
- cleanup_expired_nonces – Clean up expired pending nonces.
Attributes:
- logger –
leadr.auth.services.nonce_tasks.cleanup_expired_nonces¶
cleanup_expired_nonces()
Clean up expired pending nonces.
Deletes nonces that are:
- Status: PENDING (unused)
- Expired before current time
Used and expired nonces are kept for audit purposes.
This task is designed to be called periodically (e.g., every hour).
leadr.auth.services.nonce_tasks.logger¶
logger = logging.getLogger(__name__)
leadr.auth.services.repositories¶
API Key, Device, and Nonce repository services.
Classes:
- APIKeyRepository – API Key repository for managing API key persistence.
- DeviceRepository – Device repository for managing device persistence.
- DeviceSessionRepository – DeviceSession repository for managing device session persistence.
- NonceRepository – Nonce repository for managing nonce persistence.
leadr.auth.services.repositories.APIKeyRepository¶
Bases: BaseRepository[APIKey, APIKeyORM]
API Key repository for managing API key persistence.
Functions:
- count_active_by_account – Count active, non-deleted API keys for a given account.
- create – Create a new entity in the database.
- delete – Soft delete an entity by setting its deleted_at timestamp.
- filter – Filter API keys by account and optional criteria.
- get_by_id – Get an entity by its ID.
- get_by_prefix – Get API key by prefix, returns None if not found or soft-deleted.
- update – Update an existing entity in the database.
Attributes:
# leadr.auth.services.repositories.APIKeyRepository.SORTABLE_FIELDS¶
SORTABLE_FIELDS = {'id', 'name', 'created_at', 'updated_at'}
# leadr.auth.services.repositories.APIKeyRepository.count_active_by_account¶
count_active_by_account(account_id)
Count active, non-deleted API keys for a given account.
Parameters:
- account_id (
AccountID) – The account ID to count keys for.
Returns:
int– Number of active, non-deleted API keys for the account.
# leadr.auth.services.repositories.APIKeyRepository.create¶
create(entity)
Create a new entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity to create
Returns:
DomainEntityT– Created domain entity with refreshed data
# leadr.auth.services.repositories.APIKeyRepository.delete¶
delete(entity_id)
Soft delete an entity by setting its deleted_at timestamp.
Parameters:
- entity_id (
UUID4 | PrefixedID) – ID of entity to delete
Raises:
EntityNotFoundError– If entity is not found
# leadr.auth.services.repositories.APIKeyRepository.filter¶
filter(account_id=None, status=None, active_only=False, pagination=None, **kwargs)
Filter API keys by account and optional criteria.
Parameters:
- account_id (
AccountID | None) – Optional account ID to filter by. If None, returns all API keys (superadmin use case). Regular users should always pass account_id. - status (
APIKeyStatus | None) – Optional APIKeyStatus to filter by - active_only (
bool) – If True, only return ACTIVE keys (bool) - pagination (
PaginationParams | None) – Optional pagination parameters - **kwargs (
Any) – Additional filter parameters (reserved for future use)
Returns:
list[APIKey] | PaginatedResult[APIKey]– List of API keys if no pagination, PaginatedResult if pagination provided
Raises:
ValueError– If sort field is not in SORTABLE_FIELDSCursorValidationError– If cursor is invalid or state doesn't match
# leadr.auth.services.repositories.APIKeyRepository.get_by_id¶
get_by_id(entity_id, include_deleted=False)
Get an entity by its ID.
Parameters:
- entity_id (
UUID4 | PrefixedID) – Entity ID to retrieve - include_deleted (
bool) – If True, include soft-deleted entities. Defaults to False.
Returns:
DomainEntityT | None– Domain entity if found, None otherwise
# leadr.auth.services.repositories.APIKeyRepository.get_by_prefix¶
get_by_prefix(key_prefix)
Get API key by prefix, returns None if not found or soft-deleted.
# leadr.auth.services.repositories.APIKeyRepository.session¶
session = session
# leadr.auth.services.repositories.APIKeyRepository.update¶
update(entity)
Update an existing entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity with updated data
Returns:
DomainEntityT– Updated domain entity with refreshed data
Raises:
EntityNotFoundError– If entity is not found
leadr.auth.services.repositories.DeviceRepository¶
Bases: BaseRepository[Device, DeviceORM]
Device repository for managing device persistence.
Functions:
- create – Create a new entity in the database.
- delete – Soft delete an entity by setting its deleted_at timestamp.
- filter – Filter devices by account and optional criteria.
- get_by_game_and_fingerprint – Get device by game_id and client_fingerprint, returns None if not found or soft-deleted.
- get_by_id – Get an entity by its ID.
- update – Update an existing entity in the database.
Attributes:
# leadr.auth.services.repositories.DeviceRepository.SORTABLE_FIELDS¶
SORTABLE_FIELDS = {'id', 'platform', 'created_at', 'updated_at'}
# leadr.auth.services.repositories.DeviceRepository.create¶
create(entity)
Create a new entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity to create
Returns:
DomainEntityT– Created domain entity with refreshed data
# leadr.auth.services.repositories.DeviceRepository.delete¶
delete(entity_id)
Soft delete an entity by setting its deleted_at timestamp.
Parameters:
- entity_id (
UUID4 | PrefixedID) – ID of entity to delete
Raises:
EntityNotFoundError– If entity is not found
# leadr.auth.services.repositories.DeviceRepository.filter¶
filter(account_id=None, game_id=None, status=None, pagination=None, **kwargs)
Filter devices by account and optional criteria.
Parameters:
- account_id (
AccountID | None) – Optional account ID to filter by. If None, returns all devices (superadmin use case). Regular users should always pass account_id. - game_id (
GameID | None) – Optional game ID to filter by - status (
str | None) – Optional status string to filter by (active, banned, suspended) - pagination (
PaginationParams | None) – Optional pagination parameters - **kwargs (
Any) – Additional filter parameters (reserved for future use)
Returns:
list[Device] | PaginatedResult[Device]– List of devices if no pagination, PaginatedResult if pagination provided
Raises:
ValueError– If sort field is not in SORTABLE_FIELDSCursorValidationError– If cursor is invalid or state doesn't match
# leadr.auth.services.repositories.DeviceRepository.get_by_game_and_fingerprint¶
get_by_game_and_fingerprint(game_id, client_fingerprint)
Get device by game_id and client_fingerprint, returns None if not found or soft-deleted.
Parameters:
- game_id (
GameID) – The game ID - client_fingerprint (
str) – The client-generated SHA256 device fingerprint
Returns:
Device | None– Device if found and not deleted, None otherwise
# leadr.auth.services.repositories.DeviceRepository.get_by_id¶
get_by_id(entity_id, include_deleted=False)
Get an entity by its ID.
Parameters:
- entity_id (
UUID4 | PrefixedID) – Entity ID to retrieve - include_deleted (
bool) – If True, include soft-deleted entities. Defaults to False.
Returns:
DomainEntityT | None– Domain entity if found, None otherwise
# leadr.auth.services.repositories.DeviceRepository.session¶
session = session
# leadr.auth.services.repositories.DeviceRepository.update¶
update(entity)
Update an existing entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity with updated data
Returns:
DomainEntityT– Updated domain entity with refreshed data
Raises:
EntityNotFoundError– If entity is not found
leadr.auth.services.repositories.DeviceSessionRepository¶
Bases: BaseRepository[DeviceSession, DeviceSessionORM]
DeviceSession repository for managing device session persistence.
Functions:
- create – Create a new entity in the database.
- delete – Soft delete an entity by setting its deleted_at timestamp.
- filter – Filter sessions by account and optional criteria.
- get_by_id – Get an entity by its ID.
- get_by_refresh_token_hash – Get session by refresh token hash, returns None if not found or soft-deleted.
- get_by_token_hash – Get session by access token hash, returns None if not found or soft-deleted.
- update – Update an existing entity in the database.
Attributes:
# leadr.auth.services.repositories.DeviceSessionRepository.SORTABLE_FIELDS¶
SORTABLE_FIELDS = {'id', 'created_at', 'updated_at'}
# leadr.auth.services.repositories.DeviceSessionRepository.create¶
create(entity)
Create a new entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity to create
Returns:
DomainEntityT– Created domain entity with refreshed data
# leadr.auth.services.repositories.DeviceSessionRepository.delete¶
delete(entity_id)
Soft delete an entity by setting its deleted_at timestamp.
Parameters:
- entity_id (
UUID4 | PrefixedID) – ID of entity to delete
Raises:
EntityNotFoundError– If entity is not found
# leadr.auth.services.repositories.DeviceSessionRepository.filter¶
filter(account_id=None, device_id=None, pagination=None, **kwargs)
Filter sessions by account and optional criteria.
Note: account_id is used for multi-tenant safety via JOIN with devices table.
Parameters:
- account_id (
AccountID | None) – Optional account ID to filter by. If None, returns all sessions (superadmin use case). Regular users should always pass account_id. - device_id (
DeviceID | None) – Optional device ID to filter by - pagination (
PaginationParams | None) – Optional pagination parameters - **kwargs (
Any) – Additional filter parameters (reserved for future use)
Returns:
list[DeviceSession] | PaginatedResult[DeviceSession]– List of sessions if no pagination, PaginatedResult if pagination provided
Raises:
ValueError– If sort field is not in SORTABLE_FIELDSCursorValidationError– If cursor is invalid or state doesn't match
# leadr.auth.services.repositories.DeviceSessionRepository.get_by_id¶
get_by_id(entity_id, include_deleted=False)
Get an entity by its ID.
Parameters:
- entity_id (
UUID4 | PrefixedID) – Entity ID to retrieve - include_deleted (
bool) – If True, include soft-deleted entities. Defaults to False.
Returns:
DomainEntityT | None– Domain entity if found, None otherwise
# leadr.auth.services.repositories.DeviceSessionRepository.get_by_refresh_token_hash¶
get_by_refresh_token_hash(refresh_token_hash)
Get session by refresh token hash, returns None if not found or soft-deleted.
Parameters:
- refresh_token_hash (
str) – The hashed refresh token
Returns:
DeviceSession | None– DeviceSession if found and not deleted, None otherwise
# leadr.auth.services.repositories.DeviceSessionRepository.get_by_token_hash¶
get_by_token_hash(token_hash)
Get session by access token hash, returns None if not found or soft-deleted.
Parameters:
- token_hash (
str) – The hashed access token
Returns:
DeviceSession | None– DeviceSession if found and not deleted, None otherwise
# leadr.auth.services.repositories.DeviceSessionRepository.session¶
session = session
# leadr.auth.services.repositories.DeviceSessionRepository.update¶
update(entity)
Update an existing entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity with updated data
Returns:
DomainEntityT– Updated domain entity with refreshed data
Raises:
EntityNotFoundError– If entity is not found
leadr.auth.services.repositories.NonceRepository¶
Bases: BaseRepository[Nonce, NonceORM]
Nonce repository for managing nonce persistence.
Functions:
- cleanup_expired_nonces – Delete expired nonces older than specified time.
- create – Create a new entity in the database.
- delete – Soft delete an entity by setting its deleted_at timestamp.
- filter – Filter nonces by account and optional criteria.
- get_by_id – Get an entity by its ID.
- get_by_nonce_value – Get nonce by nonce_value, returns None if not found or soft-deleted.
- update – Update an existing entity in the database.
Attributes:
- session –
# leadr.auth.services.repositories.NonceRepository.cleanup_expired_nonces¶
cleanup_expired_nonces(before)
Delete expired nonces older than specified time.
Only deletes nonces with PENDING status. Used and expired nonces are kept for audit/debugging purposes.
Parameters:
- before (
datetime) – Delete nonces that expired before this datetime
Returns:
int– Number of nonces deleted
# leadr.auth.services.repositories.NonceRepository.create¶
create(entity)
Create a new entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity to create
Returns:
DomainEntityT– Created domain entity with refreshed data
# leadr.auth.services.repositories.NonceRepository.delete¶
delete(entity_id)
Soft delete an entity by setting its deleted_at timestamp.
Parameters:
- entity_id (
UUID4 | PrefixedID) – ID of entity to delete
Raises:
EntityNotFoundError– If entity is not found
# leadr.auth.services.repositories.NonceRepository.filter¶
filter(account_id=None, device_id=None, **kwargs)
Filter nonces by account and optional criteria.
Note: account_id is used for multi-tenant safety via JOIN with devices table.
Parameters:
- account_id (
AccountID | None) – REQUIRED - Account ID to filter by (multi-tenant safety) - device_id (
DeviceID | None) – Optional device ID to filter by
Returns:
Raises:
ValueError– If account_id is None (required for multi-tenant safety)
# leadr.auth.services.repositories.NonceRepository.get_by_id¶
get_by_id(entity_id, include_deleted=False)
Get an entity by its ID.
Parameters:
- entity_id (
UUID4 | PrefixedID) – Entity ID to retrieve - include_deleted (
bool) – If True, include soft-deleted entities. Defaults to False.
Returns:
DomainEntityT | None– Domain entity if found, None otherwise
# leadr.auth.services.repositories.NonceRepository.get_by_nonce_value¶
get_by_nonce_value(nonce_value)
Get nonce by nonce_value, returns None if not found or soft-deleted.
Parameters:
- nonce_value (
str) – The unique nonce value to search for
Returns:
Nonce | None– Nonce if found and not deleted, None otherwise
# leadr.auth.services.repositories.NonceRepository.session¶
session = session
# leadr.auth.services.repositories.NonceRepository.update¶
update(entity)
Update an existing entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity with updated data
Returns:
DomainEntityT– Updated domain entity with refreshed data
Raises:
EntityNotFoundError– If entity is not found