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.platform
platform: Mapped[str | None] = mapped_column(String, nullable=True)
# 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.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.IdentityKindEnum

Bases: str, Enum

Identity kind enum for database.

Attributes:

# leadr.auth.adapters.orm.IdentityKindEnum.CUSTOM
CUSTOM = 'CUSTOM'
# leadr.auth.adapters.orm.IdentityKindEnum.DEVICE
DEVICE = 'DEVICE'
# leadr.auth.adapters.orm.IdentityKindEnum.STEAM
STEAM = 'STEAM'
leadr.auth.adapters.orm.IdentityORM

Bases: Base

Identity ORM model.

Represents a player identity within a game in the database. Maps to the identities table with foreign keys to accounts and games. Identities are the ranking key for leaderboards.

Functions:

  • from_domain – Convert Identity domain entity to ORM model.
  • to_domain – Convert ORM model to Identity domain entity.

Attributes:

# leadr.auth.adapters.orm.IdentityORM.account_id
account_id: Mapped[UUID] = mapped_column(ForeignKey('accounts.id', ondelete='CASCADE'), nullable=False, index=True)
# leadr.auth.adapters.orm.IdentityORM.created_at
created_at: Mapped[timestamp]
# leadr.auth.adapters.orm.IdentityORM.deleted_at
deleted_at: Mapped[nullable_timestamp]
# leadr.auth.adapters.orm.IdentityORM.display_name
display_name: Mapped[str | None] = mapped_column(String, nullable=True)
# leadr.auth.adapters.orm.IdentityORM.external_key
external_key: Mapped[str] = mapped_column(String, nullable=False)
# leadr.auth.adapters.orm.IdentityORM.from_domain
from_domain(identity)

Convert Identity domain entity to ORM model.

# leadr.auth.adapters.orm.IdentityORM.game
game: Mapped[GameORM] = relationship('GameORM')
# leadr.auth.adapters.orm.IdentityORM.game_id
game_id: Mapped[UUID] = mapped_column(ForeignKey('games.id', ondelete='CASCADE'), nullable=False, index=True)
# leadr.auth.adapters.orm.IdentityORM.id
id: Mapped[uuid_pk]
# leadr.auth.adapters.orm.IdentityORM.kind
kind: Mapped[IdentityKindEnum] = mapped_column(Enum(IdentityKindEnum, name='identity_kind', native_enum=True, values_callable=(lambda x: [(e.value) for e in x])), nullable=False)
# leadr.auth.adapters.orm.IdentityORM.nonces
nonces: Mapped[list[NonceORM]] = relationship('NonceORM', cascade='all, delete-orphan')
# leadr.auth.adapters.orm.IdentityORM.sessions
sessions: Mapped[list[IdentitySessionORM]] = relationship('IdentitySessionORM', back_populates='identity', cascade='all, delete-orphan')
# leadr.auth.adapters.orm.IdentityORM.to_domain
to_domain()

Convert ORM model to Identity domain entity.

# leadr.auth.adapters.orm.IdentityORM.updated_at
updated_at: Mapped[timestamp] = mapped_column(onupdate=(func.now()))
leadr.auth.adapters.orm.IdentitySessionORM

Bases: Base

IdentitySession ORM model.

Represents an active authentication session for an identity in the database. Maps to the identity_sessions table with foreign key to identities. Sessions include both access and refresh tokens with token rotation support.

Functions:

  • from_domain – Convert IdentitySession domain entity to ORM model.
  • to_domain – Convert ORM model to IdentitySession domain entity.

Attributes:

# leadr.auth.adapters.orm.IdentitySessionORM.access_token_hash
access_token_hash: Mapped[str] = mapped_column(String, nullable=False, index=True)
# leadr.auth.adapters.orm.IdentitySessionORM.created_at
created_at: Mapped[timestamp]
# leadr.auth.adapters.orm.IdentitySessionORM.deleted_at
deleted_at: Mapped[nullable_timestamp]
# leadr.auth.adapters.orm.IdentitySessionORM.expires_at
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)
# leadr.auth.adapters.orm.IdentitySessionORM.from_domain
from_domain(session)

Convert IdentitySession domain entity to ORM model.

# leadr.auth.adapters.orm.IdentitySessionORM.id
id: Mapped[uuid_pk]
# leadr.auth.adapters.orm.IdentitySessionORM.identity
identity: Mapped[IdentityORM] = relationship('IdentityORM', back_populates='sessions')
# leadr.auth.adapters.orm.IdentitySessionORM.identity_id
identity_id: Mapped[UUID] = mapped_column(ForeignKey('identities.id', ondelete='CASCADE'), nullable=False, index=True)
# leadr.auth.adapters.orm.IdentitySessionORM.refresh_expires_at
refresh_expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True)
# leadr.auth.adapters.orm.IdentitySessionORM.refresh_token_hash
refresh_token_hash: Mapped[str] = mapped_column(String, nullable=False, index=True)
# leadr.auth.adapters.orm.IdentitySessionORM.revoked_at
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
# leadr.auth.adapters.orm.IdentitySessionORM.to_domain
to_domain()

Convert ORM model to IdentitySession domain entity.

# leadr.auth.adapters.orm.IdentitySessionORM.token_version
token_version: Mapped[int] = mapped_column(Integer, nullable=False, default=1, server_default='1')
# leadr.auth.adapters.orm.IdentitySessionORM.updated_at
updated_at: Mapped[timestamp] = mapped_column(onupdate=(func.now()))
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 identities. 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.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.identity
identity: Mapped[IdentityORM] = relationship('IdentityORM', overlaps='nonces')
# leadr.auth.adapters.orm.NonceORM.identity_id
identity_id: Mapped[UUID] = mapped_column(ForeignKey('identities.id', ondelete='CASCADE'), nullable=False, index=True)
# 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 authentication using Identity.

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 identity 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 identity authentication via access token.

Parameters:

Returns:

  • NonceResponse – NonceResponse with nonce_value and expires_at

Raises:

  • 401 – Invalid or missing access 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, identity_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(session_request, identity_service)

Start a new identity session for a game client.

This endpoint authenticates game clients and provides JWT access tokens. It is idempotent - calling multiple times for the same fingerprint updates the device record and creates a new identity session.

No authentication is required to call this endpoint (it IS the authentication).

Parameters:

  • session_request (StartSessionRequest) – Session start request with game_id and fingerprint
  • identity_service (IdentityServiceDep) – IdentityService dependency (handles device and identity creation)

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.StartSessionRequest.test_mode
test_mode: bool = Field(default=False, description='If true, session is in test mode and scores will be marked as test')
leadr.auth.api.client_schemas.StartSessionResponse

Bases: BaseModel

Response schema for starting an identity 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.display_name
display_name: str | None = Field(default=None, description='Player display name')
# 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.from_domain
from_domain(identity, access_token, refresh_token, expires_in, test_mode=False)

Convert domain entity to response model with tokens.

Parameters:

  • identity (Identity) – The domain Identity 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
  • test_mode (bool) – Whether session is in test mode

Returns:

# leadr.auth.api.client_schemas.StartSessionResponse.game_id
game_id: GameID = Field(description='ID of the game')
# leadr.auth.api.client_schemas.StartSessionResponse.identity_id
identity_id: IdentityID = Field(description='Unique identifier for the player identity')
# leadr.auth.api.client_schemas.StartSessionResponse.kind
kind: IdentityKind = Field(description='Identity type (DEVICE, STEAM, CUSTOM)')
# 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.test_mode
test_mode: bool = Field(description='Whether session is in test mode')
leadr.auth.api.device_routes

API routes for device management.

Functions:

  • get_device – Get a device by ID.
  • list_devices – List devices for an account with optional filters and pagination.
  • update_device – Update a device (change status).

Attributes:

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:

  • auth (AdminAuthContextDep) – Authentication context with user info.
  • service (DeviceServiceDep) – Injected device service dependency.
  • pagination (Annotated[PaginationParams, Depends()]) – Pagination parameters (cursor, limit, sort).
  • account_id (Annotated[AccountID | None, Query(description='Account ID filter')]) – Optional account_id query parameter (superadmins can omit to see all).
  • game_id (Annotated[GameID | None, Query(description='Filter by game ID')]) – Optional game ID to filter by.
  • device_status (Annotated[str | None, Query(alias=status, description='Filter by status')]) – Optional status to filter by (active, banned, suspended).

Returns:

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:

  • DeviceResponse – DeviceResponse with the updated device details.

Raises:

  • 403 – User does not have access to this device's account.
  • 404 – Device not found.
  • 400 – Invalid status value.
leadr.auth.api.device_schemas

API request and response models for devices.

Classes:

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

API routes for identity management.

Functions:

Attributes:

leadr.auth.api.identity_routes.get_identity
get_identity(identity_id, service, auth)

Get an identity by ID.

Parameters:

Returns:

Raises:

  • 403 – User does not have access to this identity's account.
  • 404 – Identity not found or soft-deleted.
leadr.auth.api.identity_routes.list_identities
list_identities(auth, service, pagination, account_id=None, game_id=None, kind=None)

List identities for an account with optional filters and pagination.

Returns all non-deleted identities for the specified account, with optional filtering by game or kind.

For regular users, account_id is automatically derived from their API key. For superadmins, account_id is optional - if omitted, returns identities from all accounts.

Pagination:

  • Default: 20 items per page, sorted by created_at:desc,id:asc
  • Custom sort: Use ?sort=display_name:asc,created_at:desc
  • Valid sort fields: id, display_name, kind, created_at, updated_at
  • Navigation: Use next_cursor/prev_cursor from response
Example GET /v1/identities?account_id=acc_123&game_id=game_456&kind=DEVICE&limit=50

Parameters:

  • auth (AdminAuthContextDep) – Authentication context with user info.
  • service (IdentityServiceDep) – Injected identity service dependency.
  • pagination (Annotated[PaginationParams, Depends()]) – Pagination parameters (cursor, limit, sort).
  • account_id (Annotated[AccountID | None, Query(description='Account ID filter')]) – Optional account_id query parameter (superadmins can omit to see all).
  • game_id (Annotated[GameID | None, Query(description='Filter by game ID')]) – Optional game ID to filter by.
  • kind (Annotated[str | None, Query(description='Filter by identity kind')]) – Optional kind to filter by (DEVICE, STEAM, CUSTOM).

Returns:

Raises:

  • 400 – Invalid cursor, sort field, kind, or cursor state mismatch.
  • 403 – User does not have access to the specified account.
leadr.auth.api.identity_routes.router
router = APIRouter()
leadr.auth.api.identity_routes.update_identity
update_identity(identity_id, request, service, auth)

Update an identity.

Allows updating display name or soft-deleting the identity.

Parameters:

Returns:

Raises:

  • 403 – User does not have access to this identity's account.
  • 404 – Identity not found.
leadr.auth.api.identity_schemas

API request and response models for identities.

Classes:

leadr.auth.api.identity_schemas.IdentityResponse

Bases: BaseModel

Response model for an identity.

Functions:

  • from_domain – Convert domain entity to response model.

Attributes:

# leadr.auth.api.identity_schemas.IdentityResponse.account_id
account_id: AccountID = Field(description='ID of the account this identity belongs to')
# leadr.auth.api.identity_schemas.IdentityResponse.created_at
created_at: datetime = Field(description='Timestamp when identity was created (UTC)')
# leadr.auth.api.identity_schemas.IdentityResponse.display_name
display_name: str | None = Field(default=None, description='Player display name')
# leadr.auth.api.identity_schemas.IdentityResponse.external_key
external_key: str = Field(description='External identifier (device ID, Steam ID, etc.)')
# leadr.auth.api.identity_schemas.IdentityResponse.from_domain
from_domain(identity)

Convert domain entity to response model.

Parameters:

  • identity (Identity) – The domain Identity entity to convert.

Returns:

  • IdentityResponse – IdentityResponse with all fields populated from the domain entity.
# leadr.auth.api.identity_schemas.IdentityResponse.game_id
game_id: GameID = Field(description='ID of the game this identity belongs to')
# leadr.auth.api.identity_schemas.IdentityResponse.id
id: IdentityID = Field(description='Unique identifier for the identity')
# leadr.auth.api.identity_schemas.IdentityResponse.kind
kind: str = Field(description='Identity kind: DEVICE, STEAM, or CUSTOM')
# leadr.auth.api.identity_schemas.IdentityResponse.updated_at
updated_at: datetime = Field(description='Timestamp of last update (UTC)')
leadr.auth.api.identity_schemas.IdentitySessionResponse

Bases: BaseModel

Response model for an identity session.

Functions:

  • from_domain – Convert domain entity to response model.

Attributes:

# leadr.auth.api.identity_schemas.IdentitySessionResponse.created_at
created_at: datetime = Field(description='Timestamp when session was created (UTC)')
# leadr.auth.api.identity_schemas.IdentitySessionResponse.expires_at
expires_at: datetime = Field(description='Access token expiration time (UTC)')
# leadr.auth.api.identity_schemas.IdentitySessionResponse.from_domain
from_domain(session)

Convert domain entity to response model.

Parameters:

  • session (IdentitySession) – The domain IdentitySession entity to convert.

Returns:

# leadr.auth.api.identity_schemas.IdentitySessionResponse.id
id: IdentitySessionID = Field(description='Unique identifier for the session')
# leadr.auth.api.identity_schemas.IdentitySessionResponse.identity_id
identity_id: IdentityID = Field(description='ID of the identity this session belongs to')
# leadr.auth.api.identity_schemas.IdentitySessionResponse.refresh_expires_at
refresh_expires_at: datetime = Field(description='Refresh token expiration time (UTC)')
# leadr.auth.api.identity_schemas.IdentitySessionResponse.revoked_at
revoked_at: datetime | None = Field(default=None, description='Time when session was revoked')
# leadr.auth.api.identity_schemas.IdentitySessionResponse.updated_at
updated_at: datetime = Field(description='Timestamp of last update (UTC)')
leadr.auth.api.identity_schemas.IdentityUpdateRequest

Bases: BaseModel

Request model for updating an identity.

Attributes:

# leadr.auth.api.identity_schemas.IdentityUpdateRequest.deleted
deleted: bool | None = Field(default=None, description='Set to true to soft-delete the identity')
# leadr.auth.api.identity_schemas.IdentityUpdateRequest.display_name
display_name: str | None = Field(default=None, description='Updated display name')
leadr.auth.api.identity_session_routes

API routes for identity session management.

Functions:

Attributes:

leadr.auth.api.identity_session_routes.get_identity_session
get_identity_session(session_id, service, auth)

Get an identity 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.identity_session_routes.list_identity_sessions
list_identity_sessions(auth, service, pagination, account_id=None, identity_id=None)

List identity sessions with optional filters and pagination.

Returns all non-deleted sessions, with optional filtering by account or identity.

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:desc
  • Valid sort fields: id, created_at, updated_at
  • Navigation: Use next_cursor/prev_cursor from response
Example GET /v1/identity-sessions?account_id=acc_123&identity_id=ide_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.identity_session_routes.revoke_identity_session
revoke_identity_session(session_id, service, auth)

Revoke an identity session.

Marks the session as revoked, preventing further use.

Parameters:

Returns:

Raises:

  • 403 – User does not have access to this session's account.
  • 404 – Session not found.
leadr.auth.api.identity_session_routes.router
router = APIRouter()

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 identity field.

Attributes:

leadr.auth.dependencies.AdminAuthContext
AdminAuthContext(account_id, user, api_key, identity=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).
  • identity (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 identity token.
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 - Identities 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.identity
identity: None

Get identity (always None for admin auth).

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, identity=None)

Unified authentication context for both admin and client auth.

This context provides a unified interface for both API key (admin) and identity 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).
  • identity (Identity | None) – The identity 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 identity token.
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 - Identities 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.identity
identity: Identity | None = None
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, require_superadmin=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.
  • require_superadmin (bool) – If True, only superadmin users are allowed. Returns 403 for non-superadmin users. Implies require_admin=True.

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
require_superadmin = require_superadmin
leadr.auth.dependencies.AuthContextDependency.require_superadmin_account_id
require_superadmin_account_id = require_superadmin_account_id
leadr.auth.dependencies.ClientAuthContext
ClientAuthContext(account_id, identity, user=None, api_key=None, test_mode=False)

Bases: AuthContext

Client authentication context with guaranteed identity field.

This subclass is returned by client-only authentication dependencies, providing type-safe access to identity 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 identity's game.
  • identity (Identity) – The player identity (guaranteed non-None).
  • user (None) – Always None for client auth.
  • api_key (None) – Always None for client auth.
  • test_mode (bool) – Whether this session is in test mode.

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 identity token.
leadr.auth.dependencies.ClientAuthContext.game_id
game_id: GameID

Get game ID from identity (convenience property).

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 - Identities 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.identity
identity: Identity

Get identity (guaranteed non-None for client auth).

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.test_mode
test_mode: bool

Get test_mode flag (whether session is in test mode).

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.SuperAdminAuthContextDep
SuperAdminAuthContextDep = Annotated[AdminAuthContext, Depends(require_superadmin_auth)]
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.dependencies.require_superadmin_auth
require_superadmin_auth = AuthContextDependency(require_superadmin=True)

leadr.auth.domain

Modules:

  • api_key – API Key domain model.
  • device – Device domain models for client authentication.
  • identity – Identity domain models for player identification.
  • nonce – Nonce domain entity for replay protection.
leadr.auth.domain.api_key

API Key domain model.

Classes:

  • APIKey – API Key domain entity.
  • APIKeyStatus – API Key status enumeration.
leadr.auth.domain.api_key.APIKey

Bases: Entity

API Key domain entity.

Represents an API key used to authenticate requests to the admin API. Each account can have multiple API keys for different purposes. Keys are stored hashed for security and shown only once at creation. Each API key is owned by a specific user within the account.

Functions:

  • is_expired – Check if the API key has expired.
  • is_valid – Check if the API key is valid for use.
  • record_usage – Record that the API key was used.
  • restore – Restore a soft-deleted entity.
  • revoke – Revoke the API key, preventing further use.
  • soft_delete – Mark entity as soft-deleted.
  • validate_key_prefix – Validate that key_prefix starts with 'ldr_'.

Attributes:

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

Identity domain models for player identification.

Classes:

leadr.auth.domain.identity.Identity

Bases: Entity

Identity domain entity.

Represents a player identity within a game. Identities are the ranking key for leaderboards, decoupling player identity from specific authentication mechanisms (devices, Steam accounts, etc.).

Each identity is scoped to an account and game, and is uniquely identified by the combination of kind and external_key.

Functions:

Attributes:

# leadr.auth.domain.identity.Identity.account_id
account_id: AccountID = Field(frozen=True, description='Account this identity belongs to')
# leadr.auth.domain.identity.Identity.created_at
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
# leadr.auth.domain.identity.Identity.deleted_at
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
# leadr.auth.domain.identity.Identity.display_name
display_name: str | None = Field(default=None, description='Player display name')
# leadr.auth.domain.identity.Identity.external_key
external_key: str = Field(description='External identifier (e.g., device ID, Steam ID, custom user ID)')
# leadr.auth.domain.identity.Identity.game_id
game_id: GameID = Field(frozen=True, description='Game this identity belongs to')
# leadr.auth.domain.identity.Identity.id
id: IdentityID = Field(frozen=True, default_factory=IdentityID, description='Unique identity identifier')
# leadr.auth.domain.identity.Identity.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.identity.Identity.kind
kind: IdentityKind = Field(description='Type of identity provider')
# leadr.auth.domain.identity.Identity.model_config
model_config = ConfigDict(validate_assignment=True)
# leadr.auth.domain.identity.Identity.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.identity.Identity.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.identity.Identity.update_display_name
update_display_name(name)

Update the display name.

Parameters:

  • name (str | None) – New display name, or None to clear.
# leadr.auth.domain.identity.Identity.updated_at
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
leadr.auth.domain.identity.IdentityKind

Bases: str, Enum

Identity provider type enumeration.

Attributes:

# leadr.auth.domain.identity.IdentityKind.CUSTOM
CUSTOM = 'CUSTOM'
# leadr.auth.domain.identity.IdentityKind.DEVICE
DEVICE = 'DEVICE'
# leadr.auth.domain.identity.IdentityKind.STEAM
STEAM = 'STEAM'
leadr.auth.domain.identity.IdentitySession

Bases: Entity

Identity session domain entity.

Represents an active authentication session for an identity. Sessions have an expiration time and can be revoked manually. Includes both access and refresh tokens with token rotation support.

Replaces DeviceSession with identity-based authentication.

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.identity.IdentitySession.access_token_hash
access_token_hash: str
# leadr.auth.domain.identity.IdentitySession.created_at
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
# leadr.auth.domain.identity.IdentitySession.deleted_at
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
# leadr.auth.domain.identity.IdentitySession.expires_at
expires_at: datetime
# leadr.auth.domain.identity.IdentitySession.id
id: IdentitySessionID = Field(frozen=True, default_factory=IdentitySessionID, description='Unique identity session identifier')
# leadr.auth.domain.identity.IdentitySession.identity_id
identity_id: IdentityID = Field(frozen=True, description='Identity this session belongs to')
# leadr.auth.domain.identity.IdentitySession.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.identity.IdentitySession.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.identity.IdentitySession.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.identity.IdentitySession.is_revoked
is_revoked()

Check if the session has been manually revoked.

Returns:

  • bool – True if revoked_at is set.
# leadr.auth.domain.identity.IdentitySession.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.identity.IdentitySession.model_config
model_config = ConfigDict(validate_assignment=True)
# leadr.auth.domain.identity.IdentitySession.refresh_expires_at
refresh_expires_at: datetime
# leadr.auth.domain.identity.IdentitySession.refresh_token_hash
refresh_token_hash: str
# leadr.auth.domain.identity.IdentitySession.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.identity.IdentitySession.revoke
revoke()

Revoke the session, preventing further use.

# leadr.auth.domain.identity.IdentitySession.revoked_at
revoked_at: datetime | None = None
# leadr.auth.domain.identity.IdentitySession.rotate_tokens
rotate_tokens()

Increment token version for token rotation.

Called when refreshing tokens to invalidate old refresh tokens.

# leadr.auth.domain.identity.IdentitySession.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.identity.IdentitySession.token_version
token_version: int = 1
# leadr.auth.domain.identity.IdentitySession.updated_at
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
leadr.auth.domain.nonce

Nonce domain entity for replay protection.

Classes:

  • Nonce – Request nonce for replay protection.
  • NonceStatus – Nonce status enumeration.
leadr.auth.domain.nonce.Nonce

Bases: Entity

Request nonce for replay protection.

Nonces are single-use tokens that clients must obtain before making mutating requests (POST, PATCH, DELETE). Each nonce has a short TTL (typically 60 seconds) and can only be used once.

This prevents replay attacks by ensuring that each mutating request is fresh and authorized by the server.

Functions:

  • is_expired – Check if nonce has expired.
  • is_used – Check if nonce has been used.
  • is_valid – Check if nonce is valid (not used and not expired).
  • mark_expired – Mark nonce as expired.
  • mark_used – Mark nonce as used.
  • restore – Restore a soft-deleted entity.
  • soft_delete – Mark entity as soft-deleted.

Attributes:

# 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.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.identity_id
identity_id: IdentityID = Field(description='Identity that owns this nonce')
# 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:

  • api_key_crypto – Cryptographic operations for API keys.
  • api_key_service – API Key service for managing API key operations.
  • dependencies – Auth service dependency injection factories.
  • device_service – Device authentication service.
  • device_token_crypto – Cryptographic operations for device access and refresh tokens.
  • identity_service – Identity service for player identity management.
  • nonce_service – Nonce service for managing request nonces.
  • nonce_tasks – Background tasks for nonce cleanup.
  • repositories – API Key, Device, Identity, and Nonce repository services.
leadr.auth.services.api_key_crypto

Cryptographic operations for API keys.

Functions:

  • generate_api_key – Generate a secure random API key with 'ldr_' prefix.
  • hash_api_key – Hash an API key for secure storage using HMAC-SHA256.
  • verify_api_key – Verify an API key against its stored hash.

Attributes:

leadr.auth.services.api_key_crypto.API_KEY_BYTES
API_KEY_BYTES = 32
leadr.auth.services.api_key_crypto.API_KEY_PREFIX
API_KEY_PREFIX = 'ldr_'
leadr.auth.services.api_key_crypto.generate_api_key
generate_api_key()

Generate a secure random API key with 'ldr_' prefix.

The key is generated using cryptographically secure random bytes and encoded using URL-safe base64 encoding for compatibility.

Returns:

  • str – A secure API key string starting with 'ldr_' followed 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:

  • entity_id (UUID | PrefixedID) – The ID of the entity to delete

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:

  • entity_id (UUID | PrefixedID) – The ID of the entity to retrieve

Returns:

  • DomainEntityT | None – The domain entity if found, None otherwise
# leadr.auth.services.api_key_service.APIKeyService.get_by_id_or_raise
get_by_id_or_raise(entity_id)

Get an entity by its ID or raise EntityNotFoundError.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to retrieve

Returns:

Raises:

  • EntityNotFoundError – If the entity is not found (converted to HTTP 404 by global handler)
# leadr.auth.services.api_key_service.APIKeyService.list_account_api_keys
list_account_api_keys(account_id, active_only=False)

List all API keys for an account.

This is an internal helper that returns a plain list. API keys per account are typically limited, so pagination isn't needed for internal use cases.

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)

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) – Pagination parameters (required).

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 = repository if repository is not None else 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:

  • entity_id (UUID | PrefixedID) – The ID of the entity to delete

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.IdentityServiceDep
IdentityServiceDep = Annotated[IdentityService, Depends(get_identity_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_identity_service
get_identity_service(db, device_service=Depends(get_device_service))

Get IdentityService dependency with DeviceService injected.

Parameters:

  • db (DatabaseSession) – Database session injected via dependency injection
  • device_service (DeviceService) – DeviceService injected via dependency injection

Returns:

  • IdentityService – IdentityService instance configured with the database session and device service
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:

leadr.auth.services.device_service.DeviceService
DeviceService(session)

Bases: BaseService[Device, DeviceRepository]

Service for device management.

Devices are internal lookup tables that map client fingerprints to games/accounts. For session management, use IdentityService instead.

Functions:

  • activate_device – Activate a device, allowing authentication.
  • ban_device – Ban a device, preventing further authentication.
  • delete – Soft-delete an entity.
  • get_by_id – Get an entity by its ID.
  • get_by_id_or_raise – Get an entity by its ID or raise EntityNotFoundError.
  • get_device – Get a device by its ID.
  • get_or_create_device – Get or create a device record.
  • list_all – List all non-deleted entities.
  • list_devices – List devices for an account with optional filters and pagination.
  • soft_delete – Soft-delete an entity and return it before deletion.
  • suspend_device – Suspend a device temporarily.

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:

  • entity_id (UUID | PrefixedID) – The ID of the entity to delete

Raises:

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

Get an entity by its ID.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to retrieve

Returns:

  • DomainEntityT | None – The domain entity if found, None otherwise
# leadr.auth.services.device_service.DeviceService.get_by_id_or_raise
get_by_id_or_raise(entity_id)

Get an entity by its ID or raise EntityNotFoundError.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to retrieve

Returns:

Raises:

  • EntityNotFoundError – If the entity is not found (converted to HTTP 404 by global handler)
# leadr.auth.services.device_service.DeviceService.get_device
get_device(device_id)

Get a device by its ID.

Parameters:

  • device_id (UUID) – The ID of the device to retrieve

Returns:

  • Device | None – The device if found, None otherwise
Example > > > device = await service.get_device(device_id)
# leadr.auth.services.device_service.DeviceService.get_or_create_device
get_or_create_device(game_id, client_fingerprint, platform=None, metadata=None)

Get or create a device record.

This is used as an internal lookup table to map fingerprints to games/accounts. Does NOT create sessions - use IdentityService.start_session() for that.

Parameters:

  • game_id (GameID) – Game UUID
  • client_fingerprint (str) – Client-generated SHA256 device fingerprint
  • platform (str | None) – Device platform (ios, android, etc.)
  • metadata (dict[str, Any] | None) – Additional device metadata

Returns:

  • Device (Device) – The device record (new or existing)

Raises:

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

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) – Pagination parameters (required).

Returns:

Example > > > devices = await service.list_devices( > > > ... account_id=account.id, > > > ... status="active", > > > ... pagination=pagination, > > > ... )
# leadr.auth.services.device_service.DeviceService.repository
repository = repository if repository is not None else self._create_repository(session)
# leadr.auth.services.device_service.DeviceService.session
session = session
# leadr.auth.services.device_service.DeviceService.soft_delete
soft_delete(entity_id)

Soft-delete an entity and return it before deletion.

Useful for endpoints that need to return the deleted entity in the response.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to delete

Returns:

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_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, test_mode=False, identity_id=None)

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, test_mode=False, identity_id=None)

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

Identity service for player identity management.

Classes:

  • IdentityService – Service for identity management and session handling.
leadr.auth.services.identity_service.IdentityService
IdentityService(session, device_service)

Bases: BaseService[Identity, IdentityRepository]

Service for identity management and session handling.

Functions:

Attributes:

Parameters:

  • session (AsyncSession) – SQLAlchemy async session
  • device_service (DeviceService) – DeviceService for device lookup
# leadr.auth.services.identity_service.IdentityService.delete
delete(entity_id)

Soft-delete an entity.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to delete

Raises:

# leadr.auth.services.identity_service.IdentityService.get_by_id
get_by_id(entity_id)

Get an entity by its ID.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to retrieve

Returns:

  • DomainEntityT | None – The domain entity if found, None otherwise
# leadr.auth.services.identity_service.IdentityService.get_by_id_or_raise
get_by_id_or_raise(entity_id)

Get an entity by its ID or raise EntityNotFoundError.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to retrieve

Returns:

Raises:

  • EntityNotFoundError – If the entity is not found (converted to HTTP 404 by global handler)
# leadr.auth.services.identity_service.IdentityService.get_identity
get_identity(identity_id)

Get an identity by its ID.

Parameters:

  • identity_id (IdentityID | UUID) – The ID of the identity to retrieve

Returns:

  • Identity | None – The identity if found, None otherwise
# leadr.auth.services.identity_service.IdentityService.get_identity_or_raise
get_identity_or_raise(identity_id)

Get an identity by its ID or raise EntityNotFoundError.

Parameters:

  • identity_id (IdentityID | UUID) – The ID of the identity to retrieve

Returns:

Raises:

# leadr.auth.services.identity_service.IdentityService.get_or_create_identity
get_or_create_identity(account_id, game_id, kind, external_key, display_name=None)

Get an existing identity or create a new one.

Parameters:

  • account_id (AccountID) – Account ID
  • game_id (GameID) – Game ID
  • kind (IdentityKind) – Identity kind (DEVICE, STEAM, CUSTOM)
  • external_key (str) – External identifier (e.g., device ID, Steam ID)
  • display_name (str | None) – Optional display name

Returns:

  • tuple[Identity, bool] – tuple[Identity, bool]: (identity, created) where created is True if new
# leadr.auth.services.identity_service.IdentityService.get_session
get_session(session_id)

Get an identity session by its ID.

Parameters:

Returns:

# leadr.auth.services.identity_service.IdentityService.get_session_or_raise
get_session_or_raise(session_id)

Get an identity session by its ID or raise EntityNotFoundError.

Parameters:

Returns:

Raises:

# leadr.auth.services.identity_service.IdentityService.list_all
list_all()

List all non-deleted entities.

Returns:

# leadr.auth.services.identity_service.IdentityService.list_identities
list_identities(account_id, *, game_id=None, kind=None, pagination)

List identities for an account with optional filters and pagination.

Parameters:

  • account_id (AccountID | None) – Account ID to filter by. If None, returns all identities (superadmin use case).
  • game_id (GameID | None) – Optional game ID to filter by
  • kind (IdentityKind | None) – Optional identity kind to filter by
  • pagination (PaginationParams) – Pagination parameters (required).

Returns:

# leadr.auth.services.identity_service.IdentityService.list_sessions
list_sessions(account_id, *, identity_id=None, pagination)

List identity sessions with optional filters and pagination.

Parameters:

  • account_id (AccountID | None) – Account ID to filter by. If None, returns all sessions (superadmin use case).
  • identity_id (IdentityID | None) – Optional identity ID to filter by
  • pagination (PaginationParams) – Pagination parameters (required).

Returns:

# leadr.auth.services.identity_service.IdentityService.refresh_access_token
refresh_access_token(refresh_token)

Refresh access token using a valid refresh token.

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
# leadr.auth.services.identity_service.IdentityService.repository
repository = repository if repository is not None else self._create_repository(session)
# leadr.auth.services.identity_service.IdentityService.revoke_session
revoke_session(session_id)

Revoke an identity session.

Parameters:

Returns:

Raises:

# leadr.auth.services.identity_service.IdentityService.session
session = session
# leadr.auth.services.identity_service.IdentityService.session_repo
session_repo = IdentitySessionRepository(session)
# leadr.auth.services.identity_service.IdentityService.soft_delete
soft_delete(entity_id)

Soft-delete an entity and return it before deletion.

Useful for endpoints that need to return the deleted entity in the response.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to delete

Returns:

Raises:

# leadr.auth.services.identity_service.IdentityService.start_session
start_session(game_id, client_fingerprint, platform=None, metadata=None, test_mode=False)

Start a new identity session.

Internally:

  1. Get or create Device (via DeviceService)
  2. Get or create Identity
  3. Create IdentitySession with tokens

Parameters:

  • game_id (GameID) – Game UUID
  • client_fingerprint (str) – Client-generated SHA256 device fingerprint
  • platform (str | None) – Device platform (ios, android, etc.)
  • metadata (dict[str, Any] | None) – Additional device metadata
  • test_mode (bool) – If True, session is in test mode

Returns:

  • tuple[Identity, str, str, int] – tuple[Identity, str, str, int]: (identity, access_token, refresh_token, expires_in)

Raises:

# leadr.auth.services.identity_service.IdentityService.update_identity
update_identity(identity_id, display_name=None)

Update an identity's mutable fields.

Parameters:

  • identity_id (IdentityID) – The ID of the identity to update
  • display_name (str | None) – New display name (None to clear)

Returns:

  • Identity – The updated identity

Raises:

# leadr.auth.services.identity_service.IdentityService.validate_identity_token
validate_identity_token(token)

Validate access token and return associated identity.

Validates JWT signature and expiration, checks session validity, and returns the identity.

Parameters:

  • token (str) – JWT access token

Returns:

  • Identity | None – Identity if token is valid, None otherwise
leadr.auth.services.nonce_service

Nonce service for managing request nonces.

Classes:

  • NonceService – Service for managing request nonces.
leadr.auth.services.nonce_service.NonceService

Bases: BaseService[Nonce, NonceRepository]

Service for managing request nonces.

Nonces are single-use tokens that clients must obtain before making mutating requests. This prevents replay attacks by ensuring each request is fresh and authorized by the server.

Functions:

Attributes:

# leadr.auth.services.nonce_service.NonceService.cleanup_expired_nonces
cleanup_expired_nonces(older_than_hours=24)

Clean up expired nonces older than specified hours.

Only deletes nonces with PENDING status. Used nonces are kept for audit/debugging purposes.

Parameters:

  • older_than_hours (int) – Delete nonces expired before this many hours ago (default 24)

Returns:

  • int – Number of nonces deleted
Example > > > # In background task or cron job > > > > > > deleted = await service.cleanup_expired_nonces(older_than_hours=24) > > > logger.info(f"Cleaned up {deleted} expired nonces")
# leadr.auth.services.nonce_service.NonceService.delete
delete(entity_id)

Soft-delete an entity.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to delete

Raises:

# leadr.auth.services.nonce_service.NonceService.generate_nonce
generate_nonce(identity_id, ttl_seconds=60)

Generate a fresh nonce for an identity.

Parameters:

  • identity_id (IdentityID) – Identity 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(identity_id) > > > > > > # Client includes nonce_value in leadr-client-nonce header
# leadr.auth.services.nonce_service.NonceService.get_by_id
get_by_id(entity_id)

Get an entity by its ID.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to retrieve

Returns:

  • DomainEntityT | None – The domain entity if found, None otherwise
# leadr.auth.services.nonce_service.NonceService.get_by_id_or_raise
get_by_id_or_raise(entity_id)

Get an entity by its ID or raise EntityNotFoundError.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to retrieve

Returns:

Raises:

  • EntityNotFoundError – If the entity is not found (converted to HTTP 404 by global handler)
# leadr.auth.services.nonce_service.NonceService.list_all
list_all()

List all non-deleted entities.

Returns:

# leadr.auth.services.nonce_service.NonceService.repository
repository = repository if repository is not None else self._create_repository(session)
# leadr.auth.services.nonce_service.NonceService.soft_delete
soft_delete(entity_id)

Soft-delete an entity and return it before deletion.

Useful for endpoints that need to return the deleted entity in the response.

Parameters:

  • entity_id (UUID | PrefixedID) – The ID of the entity to delete

Returns:

Raises:

# leadr.auth.services.nonce_service.NonceService.validate_and_consume_nonce
validate_and_consume_nonce(nonce_value, identity_id)

Validate nonce and mark as used (atomic operation).

Parameters:

  • nonce_value (str) – The nonce value to validate
  • identity_id (IdentityID) – Expected identity ID (must match nonce owner)

Returns:

  • bool – True if nonce was valid and consumed

Raises:

  • ValueError – If nonce is invalid (expired, already used, wrong identity, or not found)
Example > > > try: > > > ... await service.validate_and_consume_nonce(nonce_value, identity.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, Identity, 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 with pagination.
  • 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:

  • DomainEntityT – Created domain entity with refreshed data
# leadr.auth.services.repositories.APIKeyRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

# leadr.auth.services.repositories.APIKeyRepository.filter
filter(account_id=None, *, status=None, active_only=False, pagination, **kwargs)

Filter API keys by account and optional criteria with pagination.

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) – Pagination parameters (required).
  • **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:

  • entity (DomainEntityT) – Domain entity with updated data

Returns:

  • DomainEntityT – Updated domain entity with refreshed data

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 with pagination.
  • 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:

  • DomainEntityT – Created domain entity with refreshed data
# leadr.auth.services.repositories.DeviceRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

# leadr.auth.services.repositories.DeviceRepository.filter
filter(account_id=None, *, game_id=None, status=None, pagination, **kwargs)

Filter devices by account and optional criteria with pagination.

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) – Pagination parameters (required).
  • **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:

  • entity (DomainEntityT) – Domain entity with updated data

Returns:

  • DomainEntityT – Updated domain entity with refreshed data

Raises:

leadr.auth.services.repositories.IdentityRepository

Bases: BaseRepository[Identity, IdentityORM]

Identity repository for managing identity persistence.

Functions:

  • create – Create a new entity in the database.
  • delete – Soft delete an entity by setting its deleted_at timestamp.
  • filter – Filter identities by account and optional criteria with pagination.
  • get_by_external_key – Get identity by unique key combination.
  • get_by_id – Get an entity by its ID.
  • update – Update an existing entity in the database.

Attributes:

# leadr.auth.services.repositories.IdentityRepository.SORTABLE_FIELDS
SORTABLE_FIELDS = {'id', 'display_name', 'kind', 'created_at', 'updated_at'}
# leadr.auth.services.repositories.IdentityRepository.create
create(entity)

Create a new entity in the database.

Parameters:

Returns:

  • DomainEntityT – Created domain entity with refreshed data
# leadr.auth.services.repositories.IdentityRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

# leadr.auth.services.repositories.IdentityRepository.filter
filter(account_id=None, *, game_id=None, kind=None, pagination, **kwargs)

Filter identities by account and optional criteria with pagination.

Parameters:

  • account_id (AccountID | None) – Optional account ID to filter by. If None, returns all identities (superadmin use case). Regular users should always pass account_id.
  • game_id (GameID | None) – Optional game ID to filter by
  • kind (IdentityKind | None) – Optional identity kind to filter by
  • pagination (PaginationParams) – Pagination parameters (required).
  • **kwargs (Any) – Additional filter parameters (reserved for future use)

Returns:

Raises:

# leadr.auth.services.repositories.IdentityRepository.get_by_external_key
get_by_external_key(account_id, game_id, kind, external_key)

Get identity by unique key combination.

Parameters:

  • account_id (AccountID) – The account ID
  • game_id (GameID) – The game ID
  • kind (IdentityKind) – The identity kind (DEVICE, STEAM, CUSTOM)
  • external_key (str) – The external identifier

Returns:

  • Identity | None – Identity if found and not deleted, None otherwise
# leadr.auth.services.repositories.IdentityRepository.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.IdentityRepository.session
session = session
# leadr.auth.services.repositories.IdentityRepository.update
update(entity)

Update an existing entity in the database.

Parameters:

  • entity (DomainEntityT) – Domain entity with updated data

Returns:

  • DomainEntityT – Updated domain entity with refreshed data

Raises:

leadr.auth.services.repositories.IdentitySessionRepository

Bases: BaseRepository[IdentitySession, IdentitySessionORM]

IdentitySession repository for managing identity 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 with pagination.
  • 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.IdentitySessionRepository.SORTABLE_FIELDS
SORTABLE_FIELDS = {'id', 'created_at', 'updated_at'}
# leadr.auth.services.repositories.IdentitySessionRepository.create
create(entity)

Create a new entity in the database.

Parameters:

Returns:

  • DomainEntityT – Created domain entity with refreshed data
# leadr.auth.services.repositories.IdentitySessionRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

# leadr.auth.services.repositories.IdentitySessionRepository.filter
filter(account_id=None, *, identity_id=None, pagination, **kwargs)

Filter sessions by account and optional criteria with pagination.

Note: account_id is used for multi-tenant safety via JOIN with identities 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.
  • identity_id (IdentityID | None) – Optional identity ID to filter by
  • pagination (PaginationParams) – Pagination parameters (required).
  • **kwargs (Any) – Additional filter parameters (reserved for future use)

Returns:

Raises:

# leadr.auth.services.repositories.IdentitySessionRepository.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.IdentitySessionRepository.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:

  • IdentitySession | None – IdentitySession if found and not deleted, None otherwise
# leadr.auth.services.repositories.IdentitySessionRepository.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:

  • IdentitySession | None – IdentitySession if found and not deleted, None otherwise
# leadr.auth.services.repositories.IdentitySessionRepository.session
session = session
# leadr.auth.services.repositories.IdentitySessionRepository.update
update(entity)

Update an existing entity in the database.

Parameters:

  • entity (DomainEntityT) – Domain entity with updated data

Returns:

  • DomainEntityT – Updated domain entity with refreshed data

Raises:

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:

  • DomainEntityT – Created domain entity with refreshed data
# leadr.auth.services.repositories.NonceRepository.delete
delete(entity_id)

Soft delete an entity by setting its deleted_at timestamp.

Parameters:

Raises:

# leadr.auth.services.repositories.NonceRepository.filter
filter(account_id=None, identity_id=None, **kwargs)

Filter nonces by account and optional criteria.

Note: account_id is used for multi-tenant safety via JOIN with identities table.

Parameters:

  • account_id (AccountID | None) – REQUIRED - Account ID to filter by (multi-tenant safety)
  • identity_id (IdentityID | None) – Optional identity 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:

  • entity (DomainEntityT) – Domain entity with updated data

Returns:

  • DomainEntityT – Updated domain entity with refreshed data

Raises: