Skip to content

Auth

leadr.auth

Modules:

leadr.auth.adapters

Modules:

  • orm – Auth ORM models.
leadr.auth.adapters.orm

Auth ORM models.

Classes:

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:

# 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

Bases: str, Enum

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:

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

# 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

Bases: str, Enum

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:

# 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

Bases: str, Enum

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:

leadr.auth.api.api_key_routes

API routes for authentication and API key management.

Functions:

Attributes:

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:

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:

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

Parameters:

Returns:

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:

Returns:

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:

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:

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

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

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

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

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

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

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:

Returns:

Raises:

  • 401 – Invalid or expired refresh token
  • 422 – 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:

Returns:

Raises:

  • 404 – Game not found
  • 422 – Invalid request (missing required fields, invalid UUID format)
leadr.auth.api.client_schemas

API request and response models for client authentication.

Classes:

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:

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

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

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

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

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

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

Attributes:

leadr.auth.api.device_routes.get_device
get_device(device_id, service, auth)

Get a device by ID.

Parameters:

Returns:

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=50

Parameters:

Returns:

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:

Returns:

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:

leadr.auth.api.device_schemas.DeviceResponse

Bases: BaseModel

Response model for a device.

Functions:

  • from_domain – Convert domain entity to response model.

Attributes:

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

leadr.auth.api.device_session_routes.get_session
get_session(session_id, service, auth)

Get a device session by ID.

Parameters:

Returns:

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=50

Parameters:

Returns:

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:

Returns:

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:

leadr.auth.api.device_session_schemas.DeviceSessionResponse

Bases: BaseModel

Response model for device session.

Functions:

  • from_domain – Convert domain entity to API response.

Attributes:

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

Attributes:

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:

  1. Check if any superadmin user already exists
  2. If not, create:
  3. A system account (configured via SUPERADMIN_ACCOUNT_NAME/SLUG)
  4. A superadmin user (configured via SUPERADMIN_EMAIL/DISPLAY_NAME)
  5. 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:

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:

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 account
For client auth - Devices only have access to their game's account

Parameters:

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

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 account
For client auth - Devices only have access to their game's account

Parameters:

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

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 account
For client auth - Devices only have access to their game's account

Parameters:

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

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:

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

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:

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

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

Attributes:

# 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

Bases: str, Enum

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:

leadr.auth.services.api_key_crypto

Cryptographic operations for API keys.

Functions:

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 by
  • str – URL-safe random characters (alphanumeric, hyphen, underscore).
Example > > > key = generate_api_key() > > > key.startswith('ldr\_') > > > True > > > len(key) > 36 > > > True
leadr.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:

  • key (str) – The API key to hash.
  • secret (str) – Server-side secret for additional security.

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) > > > 64
leadr.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) > > > False
leadr.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:

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:

Raises:

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

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:

Returns:

Raises:

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

  • list[APIKey] – List of APIKey domain entities.
# leadr.auth.services.api_key_service.APIKeyService.list_all
list_all()

List all non-deleted entities.

Returns:

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

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

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

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

Returns:

Raises:

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

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

  1. Extracts prefix and looks up key in database
  2. Verifies the hash matches
  3. Checks if key is active (not revoked)
  4. Checks if key is not expired
  5. 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:

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:

Attributes:

Parameters:

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

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:

Example > > > device = await service.ban_device(device_id)
# leadr.auth.services.device_service.DeviceService.delete
delete(entity_id)

Soft-delete an entity.

Parameters:

Raises:

# leadr.auth.services.device_service.DeviceService.get_by_id
get_by_id(entity_id)

Get an entity by its ID.

Parameters:

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:

Returns:

Raises:

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

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:

Returns:

Raises:

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:

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

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:

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:

Returns:

Raises:

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:

Returns:

Raises:

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

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

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:

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(".") > > > 2
leadr.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(".") > > > 2
leadr.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:

  • token (str) – The JWT token to hash
  • secret (str) – Server-side secret for additional security

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) > > > 64
leadr.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:

  • token (str) – JWT access token to validate
  • secret (str) – Server-side secret for JWT verification

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:

  • token (str) – JWT refresh token to validate
  • secret (str) – Server-side secret for JWT verification

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 > > > 1
leadr.auth.services.nonce_service

Nonce service for managing request nonces.

Classes:

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:

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:

Raises:

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

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:

Returns:

Raises:

# leadr.auth.services.nonce_service.NonceService.list_all
list_all()

List all non-deleted entities.

Returns:

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

Returns:

Raises:

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

Attributes:

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:

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:

Returns:

# leadr.auth.services.repositories.APIKeyRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

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

Raises:

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

Returns:

Raises:

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:

Returns:

# leadr.auth.services.repositories.DeviceRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

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

Raises:

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

Returns:

Raises:

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:

Returns:

# leadr.auth.services.repositories.DeviceSessionRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

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

Raises:

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

Returns:

Raises:

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:

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

Returns:

# leadr.auth.services.repositories.NonceRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

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

  • list[Nonce] – List of nonces matching the filter criteria

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:

Returns:

Raises: