Common
leadr.common¶
Modules:
- api – API utilities and exception handlers.
- background_tasks – Background task scheduler using asyncio.
- database – Database connection and session management.
- dependencies – Shared FastAPI dependencies for the application.
- domain –
- geoip – GeoIP service for IP address geolocation using MaxMind databases.
- orm – Common ORM base classes and utilities.
- repositories – Base repository abstraction for common CRUD operations.
- services – Base service abstraction for common business logic patterns.
- utils – Common utility functions.
leadr.common.api¶
API utilities and exception handlers.
Modules:
- exceptions – Global exception handlers for API layer.
- hooks – Request hooks for cloud/enterprise extensibility.
- pagination – API pagination models and dependencies.
leadr.common.api.exceptions¶
Global exception handlers for API layer.
Functions:
- catchall_exception_handler – Convert all unhandled Exceptions to a 500 HTTP response with
- entity_not_found_handler – Convert EntityNotFoundError to 404 HTTP response.
- http_exception_handler – Convert all FastAPI HTTPExceptions to ensure our response envelope.
- validation_error_handler – Convert RequestValidationError to 422 HTTP response with our
Attributes:
- logger –
leadr.common.api.exceptions.catchall_exception_handler¶
catchall_exception_handler(request, exc)
Convert all unhandled Exceptions to a 500 HTTP response with our response envelope.
Parameters:
Returns:
JSONResponse– JSONResponse with 500 status and error detail
leadr.common.api.exceptions.entity_not_found_handler¶
entity_not_found_handler(request, exc)
Convert EntityNotFoundError to 404 HTTP response.
Parameters:
- request (
Request) – The incoming request - exc (
EntityNotFoundError) – The domain exception
Returns:
JSONResponse– JSONResponse with 404 status and error detail
leadr.common.api.exceptions.http_exception_handler¶
http_exception_handler(request, exc)
Convert all FastAPI HTTPExceptions to ensure our response envelope.
Parameters:
- request (
Request) – The incoming request - exc (
HTTPException) – The exception
Returns:
JSONResponse– JSONResponse with HTTP status code and error detail
leadr.common.api.exceptions.logger¶
logger = logging.getLogger(__name__)
leadr.common.api.exceptions.validation_error_handler¶
validation_error_handler(request, exc)
Convert RequestValidationError to 422 HTTP response with our error response envelope.
Parameters:
- request (
Request) – The incoming request - exc (
RequestValidationError) – The validation exception
Returns:
JSONResponse– JSONResponse with 422 status and list of validation errors
leadr.common.api.hooks¶
Request hooks for cloud/enterprise extensibility.
These hooks are no-ops in OSS but can be overridden via FastAPI dependency_overrides.
Classes:
- Hook – Generic hook type for resource lifecycle events.
- UpdateHook – Hook type for update operations that need the resource's account_id.
Functions:
- get_post_create_board_hook – Get the post-create board hook. Override in cloud.
- get_post_create_game_hook – Get the post-create game hook. Override in cloud.
- get_post_create_score_hook – Get the post-create score hook. Override in cloud.
- get_pre_create_board_hook – Get the pre-create board hook. Override in cloud.
- get_pre_create_board_template_hook – Get the pre-create board template hook. Override in cloud.
- get_pre_create_game_hook – Get the pre-create game hook. Override in cloud.
- get_pre_create_score_hook – Get the pre-create score hook. Override in cloud.
- get_pre_update_board_template_hook – Get the pre-update board template hook. Override in cloud.
- get_rate_limit_hook – Get the rate limit hook. Override in cloud.
- noop_post_create_board – No-op post-create board hook.
- noop_post_create_game – No-op post-create game hook.
- noop_post_create_score – No-op post-create score hook.
- noop_pre_create_board – No-op pre-create board hook.
- noop_pre_create_board_template – No-op pre-create board template hook.
- noop_pre_create_game – No-op pre-create game hook.
- noop_pre_create_score – No-op pre-create score hook.
- noop_pre_update_board_template – No-op pre-update board template hook.
- noop_rate_limit_check – No-op rate limit hook.
- require_rate_limit_check – Rate limit dependency for router-level application.
Attributes:
- AuthT –
- PostCreateBoardHook (
TypeAlias) – - PostCreateBoardHookDep –
- PostCreateGameHook (
TypeAlias) – - PostCreateGameHookDep –
- PostCreateScoreHook (
TypeAlias) – - PostCreateScoreHookDep –
- PreCreateBoardHook (
TypeAlias) – - PreCreateBoardHookDep –
- PreCreateBoardTemplateHook (
TypeAlias) – - PreCreateBoardTemplateHookDep –
- PreCreateGameHook (
TypeAlias) – - PreCreateGameHookDep –
- PreCreateScoreHook (
TypeAlias) – - PreCreateScoreHookDep –
- PreUpdateBoardTemplateHook (
TypeAlias) – - PreUpdateBoardTemplateHookDep –
- RateLimitHook –
- RequestT –
leadr.common.api.hooks.AuthT¶
AuthT = TypeVar('AuthT', contravariant=True)
leadr.common.api.hooks.Hook¶
Bases: Protocol[RequestT, AuthT]
Generic hook type for resource lifecycle events.
All hooks follow the signature: (request, auth, background_tasks) -> None
- request: The Pydantic request schema for the operation
- auth: The authentication context (AdminAuthContext or ClientAuthContext)
- background_tasks: FastAPI BackgroundTasks for scheduling async work
leadr.common.api.hooks.PostCreateBoardHook¶
PostCreateBoardHook: TypeAlias = Hook[BoardCreateRequest, AdminAuthContext]
leadr.common.api.hooks.PostCreateBoardHookDep¶
PostCreateBoardHookDep = Annotated[PostCreateBoardHook, Depends(get_post_create_board_hook)]
leadr.common.api.hooks.PostCreateGameHook¶
PostCreateGameHook: TypeAlias = Hook[GameCreateRequest, AdminAuthContext]
leadr.common.api.hooks.PostCreateGameHookDep¶
PostCreateGameHookDep = Annotated[PostCreateGameHook, Depends(get_post_create_game_hook)]
leadr.common.api.hooks.PostCreateScoreHook¶
PostCreateScoreHook: TypeAlias = Hook[ScoreClientCreateRequest, ClientAuthContext]
leadr.common.api.hooks.PostCreateScoreHookDep¶
PostCreateScoreHookDep = Annotated[PostCreateScoreHook, Depends(get_post_create_score_hook)]
leadr.common.api.hooks.PreCreateBoardHook¶
PreCreateBoardHook: TypeAlias = Hook[BoardCreateRequest, AdminAuthContext]
leadr.common.api.hooks.PreCreateBoardHookDep¶
PreCreateBoardHookDep = Annotated[PreCreateBoardHook, Depends(get_pre_create_board_hook)]
leadr.common.api.hooks.PreCreateBoardTemplateHook¶
PreCreateBoardTemplateHook: TypeAlias = Hook[BoardTemplateCreateRequest, AdminAuthContext]
leadr.common.api.hooks.PreCreateBoardTemplateHookDep¶
PreCreateBoardTemplateHookDep = Annotated[PreCreateBoardTemplateHook, Depends(get_pre_create_board_template_hook)]
leadr.common.api.hooks.PreCreateGameHook¶
PreCreateGameHook: TypeAlias = Hook[GameCreateRequest, AdminAuthContext]
leadr.common.api.hooks.PreCreateGameHookDep¶
PreCreateGameHookDep = Annotated[PreCreateGameHook, Depends(get_pre_create_game_hook)]
leadr.common.api.hooks.PreCreateScoreHook¶
PreCreateScoreHook: TypeAlias = Hook[ScoreClientCreateRequest, ClientAuthContext]
leadr.common.api.hooks.PreCreateScoreHookDep¶
PreCreateScoreHookDep = Annotated[PreCreateScoreHook, Depends(get_pre_create_score_hook)]
leadr.common.api.hooks.PreUpdateBoardTemplateHook¶
PreUpdateBoardTemplateHook: TypeAlias = UpdateHook[BoardTemplateUpdateRequest, AdminAuthContext]
leadr.common.api.hooks.PreUpdateBoardTemplateHookDep¶
PreUpdateBoardTemplateHookDep = Annotated[PreUpdateBoardTemplateHook, Depends(get_pre_update_board_template_hook)]
leadr.common.api.hooks.RateLimitHook¶
RateLimitHook = Callable[[Request], Awaitable[None]]
leadr.common.api.hooks.RequestT¶
RequestT = TypeVar('RequestT', contravariant=True)
leadr.common.api.hooks.UpdateHook¶
Bases: Protocol[RequestT, AuthT]
Hook type for update operations that need the resource's account_id.
Update hooks receive account_id separately since it comes from the existing resource, not the update request body.
leadr.common.api.hooks.get_post_create_board_hook¶
get_post_create_board_hook()
Get the post-create board hook. Override in cloud.
leadr.common.api.hooks.get_post_create_game_hook¶
get_post_create_game_hook()
Get the post-create game hook. Override in cloud.
leadr.common.api.hooks.get_post_create_score_hook¶
get_post_create_score_hook()
Get the post-create score hook. Override in cloud.
leadr.common.api.hooks.get_pre_create_board_hook¶
get_pre_create_board_hook()
Get the pre-create board hook. Override in cloud.
leadr.common.api.hooks.get_pre_create_board_template_hook¶
get_pre_create_board_template_hook()
Get the pre-create board template hook. Override in cloud.
leadr.common.api.hooks.get_pre_create_game_hook¶
get_pre_create_game_hook()
Get the pre-create game hook. Override in cloud.
leadr.common.api.hooks.get_pre_create_score_hook¶
get_pre_create_score_hook()
Get the pre-create score hook. Override in cloud.
leadr.common.api.hooks.get_pre_update_board_template_hook¶
get_pre_update_board_template_hook()
Get the pre-update board template hook. Override in cloud.
leadr.common.api.hooks.get_rate_limit_hook¶
get_rate_limit_hook()
Get the rate limit hook. Override in cloud.
leadr.common.api.hooks.noop_post_create_board¶
noop_post_create_board(request, auth, background_tasks)
No-op post-create board hook.
leadr.common.api.hooks.noop_post_create_game¶
noop_post_create_game(request, auth, background_tasks)
No-op post-create game hook.
leadr.common.api.hooks.noop_post_create_score¶
noop_post_create_score(request, auth, background_tasks)
No-op post-create score hook.
leadr.common.api.hooks.noop_pre_create_board¶
noop_pre_create_board(request, auth, background_tasks)
No-op pre-create board hook.
leadr.common.api.hooks.noop_pre_create_board_template¶
noop_pre_create_board_template(request, auth, background_tasks)
No-op pre-create board template hook.
leadr.common.api.hooks.noop_pre_create_game¶
noop_pre_create_game(request, auth, background_tasks)
No-op pre-create game hook.
leadr.common.api.hooks.noop_pre_create_score¶
noop_pre_create_score(request, auth, background_tasks)
No-op pre-create score hook.
leadr.common.api.hooks.noop_pre_update_board_template¶
noop_pre_update_board_template(account_id, request, auth, background_tasks)
No-op pre-update board template hook.
leadr.common.api.hooks.noop_rate_limit_check¶
noop_rate_limit_check(request)
No-op rate limit hook.
leadr.common.api.hooks.require_rate_limit_check¶
require_rate_limit_check(request, hook)
Rate limit dependency for router-level application.
leadr.common.api.pagination¶
API pagination models and dependencies.
Classes:
- PaginatedResponse – Generic paginated response wrapper.
- PaginationMeta – Pagination metadata in API responses.
- PaginationParams – FastAPI dependency for parsing pagination query parameters.
Attributes:
leadr.common.api.pagination.DomainT¶
DomainT = TypeVar('DomainT')
leadr.common.api.pagination.PaginatedResponse¶
Generic paginated response wrapper.
Wraps a list of items with pagination metadata.
Functions:
- from_paginated_result – Create a PaginatedResponse from a PaginatedResult.
Attributes:
- data (
list[T]) – - model_config –
- pagination (
PaginationMeta) –
# leadr.common.api.pagination.PaginatedResponse.data¶
data: list[T] = Field(description='List of items in this page')
# leadr.common.api.pagination.PaginatedResponse.from_paginated_result¶
from_paginated_result(result, pagination, filters, response_model)
Create a PaginatedResponse from a PaginatedResult.
This factory method abstracts away cursor construction, converting a repository-layer PaginatedResult into an API-layer PaginatedResponse with encoded cursors.
Parameters:
- result (
PaginatedResult[DomainT]) – The paginated result from the repository layer - pagination (
PaginationParams) – The pagination parameters from the request - filters (
dict[str, Any]) – Dict of active filters to include in cursors (e.g., {"game_id": "123"}) - response_model (
type[ResponseT]) – The response model class with a from_domain() method
Returns:
PaginatedResponse[ResponseT]– A fully constructed PaginatedResponse with encoded cursors and converted items
Example
> > > return PaginatedResponse.from_paginated_result( > > > ... result=result, > > > ... pagination=pagination, > > > ... filters={"game_id": str(game_id)} if game_id else {}, > > > ... response_model=ScoreResponse, > > > ... )# leadr.common.api.pagination.PaginatedResponse.model_config¶
model_config = ConfigDict(json_schema_extra={'example': {'data': [{'id': 'scr_123', 'value': 1000}], 'pagination': {'next_cursor': 'eyJwdiI6WzEwMDAsMTIzXX0=', 'prev_cursor': None, 'has_next': True, 'has_prev': False, 'count': 20}}})
# leadr.common.api.pagination.PaginatedResponse.pagination¶
pagination: PaginationMeta = Field(description='Pagination metadata')
leadr.common.api.pagination.PaginationMeta¶
Bases: BaseModel
Pagination metadata in API responses.
Attributes:
- count (
int) – - has_next (
bool) – - has_prev (
bool) – - next_cursor (
str | None) – - prev_cursor (
str | None) –
# leadr.common.api.pagination.PaginationMeta.count¶
count: int = Field(description='Number of items in this page')
# leadr.common.api.pagination.PaginationMeta.has_next¶
has_next: bool = Field(description='Whether there are more results after this page')
# leadr.common.api.pagination.PaginationMeta.has_prev¶
has_prev: bool = Field(description='Whether there are results before this page')
# leadr.common.api.pagination.PaginationMeta.next_cursor¶
next_cursor: str | None = Field(None, description='Cursor for the next page of results')
# leadr.common.api.pagination.PaginationMeta.prev_cursor¶
prev_cursor: str | None = Field(None, description='Cursor for the previous page of results')
leadr.common.api.pagination.PaginationParams¶
PaginationParams(cursor=Query(None, description='Pagination cursor for navigating results'), limit=Query(20, ge=1, le=100, description='Number of items per page (1-100)'), sort=Query(None, description="Sort specification (e.g., 'value:desc,created_at:asc')"))
FastAPI dependency for parsing pagination query parameters.
Parses cursor, limit, and sort parameters from the query string. Always appends 'id:asc' to sort specification for stable sorting.
Functions:
- decode_cursor – Decode the cursor if present.
- has_cursor – Check if a cursor is present.
Attributes:
- cursor_str –
- limit –
- sort_spec –
Parameters:
- cursor (
str | None) – Optional base64-encoded cursor string - limit (
int) – Page size (1-100, default 20) - sort (
str | None) – Comma-separated sort spec (e.g., "score:desc,created_at:asc")
# leadr.common.api.pagination.PaginationParams.cursor_str¶
cursor_str = cursor
# leadr.common.api.pagination.PaginationParams.decode_cursor¶
decode_cursor()
Decode the cursor if present.
Returns:
Cursor | None– Decoded Cursor object or None if no cursor
Raises:
CursorValidationError– If cursor is invalid
# leadr.common.api.pagination.PaginationParams.has_cursor¶
has_cursor()
Check if a cursor is present.
# leadr.common.api.pagination.PaginationParams.limit¶
limit = limit
# leadr.common.api.pagination.PaginationParams.sort_spec¶
sort_spec = self._parse_sort(sort)
leadr.common.api.pagination.ResponseT¶
ResponseT = TypeVar('ResponseT', bound=BaseModel)
leadr.common.api.pagination.T¶
T = TypeVar('T')
leadr.common.background_tasks¶
Background task scheduler using asyncio.
Provides a simple background task scheduler that runs periodic tasks within the FastAPI application process.
Classes:
- BackgroundTaskScheduler – Manages periodic background tasks using asyncio.
Functions:
- get_scheduler – Get the global scheduler instance.
Attributes:
- logger –
leadr.common.background_tasks.BackgroundTaskScheduler¶
BackgroundTaskScheduler()
Manages periodic background tasks using asyncio.
Tasks run in the same process as the FastAPI application, making them easy to test and deploy without additional infrastructure.
Example
> > > scheduler = BackgroundTaskScheduler() > > > async def my_task(): > > > ... print("Task running") > > > scheduler.add_task("my-task", my_task, interval_seconds=60) > > > await scheduler.start()Functions:
- add_task – Register a periodic task.
- start – Start all registered tasks.
- stop – Stop all running tasks gracefully.
Attributes:
leadr.common.background_tasks.BackgroundTaskScheduler.add_task¶
add_task(name, func, interval_seconds)
Register a periodic task.
Parameters:
- name (
str) – Unique identifier for the task. - func (
Callable[[], Awaitable[None]]) – Async function to call periodically. - interval_seconds (
int) – How often to run the task (in seconds).
Raises:
ValueError– If task with the same name already exists.
leadr.common.background_tasks.BackgroundTaskScheduler.running¶
running = False
leadr.common.background_tasks.BackgroundTaskScheduler.start¶
start()
Start all registered tasks.
This method starts all background task loops concurrently. It returns immediately after starting the tasks.
leadr.common.background_tasks.BackgroundTaskScheduler.stop¶
stop()
Stop all running tasks gracefully.
Waits for currently executing tasks to complete before stopping.
leadr.common.background_tasks.BackgroundTaskScheduler.tasks¶
tasks: dict[str, dict[str, Any]] = {}
leadr.common.background_tasks.get_scheduler¶
get_scheduler()
Get the global scheduler instance.
Returns:
BackgroundTaskScheduler– The singleton BackgroundTaskScheduler instance.
leadr.common.background_tasks.logger¶
logger = logging.getLogger(__name__)
leadr.common.database¶
Database connection and session management.
Functions:
- build_database_url – Build async database URL from settings.
- build_direct_database_url – Build async database URL for direct connections (migrations).
- get_db – FastAPI dependency for async database session.
Attributes:
- async_session_factory –
- engine –
leadr.common.database.async_session_factory¶
async_session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
leadr.common.database.build_database_url¶
build_database_url()
Build async database URL from settings.
leadr.common.database.build_direct_database_url¶
build_direct_database_url()
Build async database URL for direct connections (migrations).
Uses DB_HOST_DIRECT if set, otherwise falls back to DB_HOST. For Neon, DB_HOST_DIRECT should be the non-pooler endpoint to avoid connecting through PgBouncer during migrations.
leadr.common.database.engine¶
engine = create_async_engine(build_database_url(), connect_args=(_get_connect_args()), poolclass=(_get_pool_class()), **(_get_pool_options()), echo=(settings.DB_ECHO))
leadr.common.database.get_db¶
get_db()
FastAPI dependency for async database session.
The session is yielded to the caller, who is responsible for committing transactions explicitly. The context manager automatically handles cleanup and rollback on exceptions.
leadr.common.dependencies¶
Shared FastAPI dependencies for the application.
Attributes:
- DatabaseSession –
leadr.common.dependencies.DatabaseSession¶
DatabaseSession = Annotated[AsyncSession, Depends(get_db)]
leadr.common.domain¶
Modules:
- cursor – Cursor-based pagination implementation.
- exceptions – Domain-specific exceptions for LEADR.
- ids – Prefixed ID types for entity identification.
- models – Common domain models and value objects.
- pagination – Pagination domain models.
- pagination_result – Pagination result from repository layer.
leadr.common.domain.cursor¶
Cursor-based pagination implementation.
Classes:
- Cursor – Opaque cursor for pagination that encodes position, sort, filters, and direction.
- CursorValidationError – Raised when cursor validation fails.
Attributes:
- logger –
leadr.common.domain.cursor.Cursor¶
Cursor(position, sort_fields, filters, direction)
Opaque cursor for pagination that encodes position, sort, filters, and direction.
The cursor ensures that pagination state (sort order, filters) remains consistent across page requests. If the client changes sort/filter parameters while using a cursor, validation will fail.
Functions:
- decode – Decode cursor from base64 string.
- encode – Encode cursor to base64 string.
- validate_state – Validate that current query state matches cursor state.
Attributes:
- direction –
- filters –
- position –
- sort_fields –
Parameters:
- position (
CursorPosition) – The position in the result set (values + entity_id) - sort_fields (
list[SortField]) – List of sort fields that were applied - filters (
dict[str, Any]) – Dictionary of filter parameters that were applied - direction (
PaginationDirection) – Direction of pagination (forward or backward)
# leadr.common.domain.cursor.Cursor.decode¶
decode(cursor_str)
Decode cursor from base64 string.
Parameters:
- cursor_str (
str) – Base64-encoded cursor string
Returns:
Cursor– Decoded Cursor object
Raises:
CursorValidationError– If cursor is invalid or malformed
# leadr.common.domain.cursor.Cursor.direction¶
direction = direction
# leadr.common.domain.cursor.Cursor.encode¶
encode()
Encode cursor to base64 string.
Returns:
str– Base64-encoded JSON string representing the cursor state
# leadr.common.domain.cursor.Cursor.filters¶
filters = filters
# leadr.common.domain.cursor.Cursor.position¶
position = position
# leadr.common.domain.cursor.Cursor.sort_fields¶
sort_fields = sort_fields
# leadr.common.domain.cursor.Cursor.validate_state¶
validate_state(sort_fields, filters)
Validate that current query state matches cursor state.
Parameters:
- sort_fields (
list[SortField]) – Current sort fields - filters (
dict[str, Any]) – Current filter parameters
Raises:
CursorValidationError– If state doesn't match
leadr.common.domain.cursor.CursorValidationError¶
Bases: ValueError
Raised when cursor validation fails.
leadr.common.domain.cursor.logger¶
logger = logging.getLogger(__name__)
leadr.common.domain.exceptions¶
Domain-specific exceptions for LEADR.
Classes:
- DomainError – Base exception for all domain-level errors.
- EntityNotFoundError – Raised when an entity cannot be found in the repository.
- InvalidEntityStateError – Raised when an entity is in an invalid state for the requested operation.
- ValidationError – Raised when entity validation fails.
leadr.common.domain.exceptions.DomainError¶
DomainError(message)
Bases: Exception
Base exception for all domain-level errors.
All custom domain exceptions should inherit from this base class. This allows catching all domain errors with a single except clause.
Parameters:
- message (
str) – Human-readable error message.
Example
> > > try: > > > ... raise DomainError("Something went wrong") > > > ... except DomainError as e: > > > ... print(e.message)Attributes:
- message –
# leadr.common.domain.exceptions.DomainError.message¶
message = message
leadr.common.domain.exceptions.EntityNotFoundError¶
EntityNotFoundError(entity_type, entity_id)
Bases: DomainError
Raised when an entity cannot be found in the repository.
This exception is raised by repository queries when an entity with the specified ID does not exist in the database.
Parameters:
- entity_type (
str) – Name of the entity type (e.g., "Account", "User"). - entity_id (
str) – The ID that was not found.
Example
> > > raise EntityNotFoundError("Account", "123e4567-e89b-12d3-a456-426614174000") > > > EntityNotFoundError: Account not found: 123e4567-e89b-12d3-a456-426614174000Attributes:
- entity_id –
- entity_type –
- message –
# leadr.common.domain.exceptions.EntityNotFoundError.entity_id¶
entity_id = entity_id
# leadr.common.domain.exceptions.EntityNotFoundError.entity_type¶
entity_type = entity_type
# leadr.common.domain.exceptions.EntityNotFoundError.message¶
message = message
leadr.common.domain.exceptions.InvalidEntityStateError¶
InvalidEntityStateError(entity_type, reason)
Bases: DomainError
Raised when an entity is in an invalid state for the requested operation.
Use this exception when business rules prevent an operation due to the current state of an entity. For example, activating an already-active account, or modifying a deleted entity.
Parameters:
- entity_type (
str) – Name of the entity type (e.g., "Account", "User"). - reason (
str) – Explanation of why the state is invalid.
Example
> > > raise InvalidEntityStateError("Account", "Cannot activate already active account") > > > InvalidEntityStateError: Invalid Account state: Cannot activate already active accountAttributes:
- entity_type –
- message –
- reason –
# leadr.common.domain.exceptions.InvalidEntityStateError.entity_type¶
entity_type = entity_type
# leadr.common.domain.exceptions.InvalidEntityStateError.message¶
message = message
# leadr.common.domain.exceptions.InvalidEntityStateError.reason¶
reason = reason
leadr.common.domain.exceptions.ValidationError¶
ValidationError(entity_type, field, reason)
Bases: DomainError
Raised when entity validation fails.
Use this exception when field-level validation fails, such as invalid format, out-of-range values, or constraint violations.
Parameters:
- entity_type (
str) – Name of the entity type (e.g., "Account", "User"). - field (
str) – Name of the field that failed validation. - reason (
str) – Explanation of why validation failed.
Example
> > > raise ValidationError("Account", "slug", "Must be lowercase alphanumeric") > > > ValidationError: Account.slug: Must be lowercase alphanumericAttributes:
- entity_type –
- field –
- message –
- reason –
# leadr.common.domain.exceptions.ValidationError.entity_type¶
entity_type = entity_type
# leadr.common.domain.exceptions.ValidationError.field¶
field = field
# leadr.common.domain.exceptions.ValidationError.message¶
message = message
# leadr.common.domain.exceptions.ValidationError.reason¶
reason = reason
leadr.common.domain.ids¶
Prefixed ID types for entity identification.
Classes:
- APIKeyID – API key entity identifier.
- AccountID – Account entity identifier.
- BoardID – Board entity identifier.
- BoardRatioConfigID – Board ratio config entity identifier.
- BoardStateID – Board state entity identifier.
- BoardTemplateID – Board template entity identifier.
- DeviceID – Device entity identifier.
- DeviceSessionID – Device session entity identifier.
- EmailID – Email entity identifier.
- GameID – Game entity identifier.
- IdentityID – Identity entity identifier.
- IdentitySessionID – Identity session entity identifier.
- JamCodeID – Jam Code entity identifier.
- JamCodeRedemptionID – Jam Code Redemption entity identifier.
- NonceID – Nonce entity identifier.
- PrefixedID – Base class for entity IDs with type prefixes.
- RunEntryID – Run entry entity identifier.
- ScoreEventID – Score event entity identifier.
- ScoreFlagID – Score flag entity identifier.
- ScoreID – Score entity identifier.
- ScoreSubmissionMetaID – Score submission metadata entity identifier.
- UserID – User entity identifier.
leadr.common.domain.ids.APIKeyID¶
Bases: PrefixedID
API key entity identifier.
Attributes:
# leadr.common.domain.ids.APIKeyID.prefix¶
prefix = 'key'
# leadr.common.domain.ids.APIKeyID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.AccountID¶
Bases: PrefixedID
Account entity identifier.
Attributes:
# leadr.common.domain.ids.AccountID.prefix¶
prefix = 'acc'
# leadr.common.domain.ids.AccountID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.BoardID¶
Bases: PrefixedID
Board entity identifier.
Attributes:
# leadr.common.domain.ids.BoardID.prefix¶
prefix = 'brd'
# leadr.common.domain.ids.BoardID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.BoardRatioConfigID¶
Bases: PrefixedID
Board ratio config entity identifier.
Attributes:
# leadr.common.domain.ids.BoardRatioConfigID.prefix¶
prefix = 'brc'
# leadr.common.domain.ids.BoardRatioConfigID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.BoardStateID¶
Bases: PrefixedID
Board state entity identifier.
Attributes:
# leadr.common.domain.ids.BoardStateID.prefix¶
prefix = 'bst'
# leadr.common.domain.ids.BoardStateID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.BoardTemplateID¶
Bases: PrefixedID
Board template entity identifier.
Attributes:
# leadr.common.domain.ids.BoardTemplateID.prefix¶
prefix = 'tpl'
# leadr.common.domain.ids.BoardTemplateID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.DeviceID¶
Bases: PrefixedID
Device entity identifier.
Attributes:
# leadr.common.domain.ids.DeviceID.prefix¶
prefix = 'dev'
# leadr.common.domain.ids.DeviceID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.DeviceSessionID¶
Bases: PrefixedID
Device session entity identifier.
Deprecated: Use IdentitySessionID instead. Will be removed in cleanup phase.
Attributes:
# leadr.common.domain.ids.DeviceSessionID.prefix¶
prefix = 'ses'
# leadr.common.domain.ids.DeviceSessionID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.EmailID¶
Bases: PrefixedID
Email entity identifier.
Attributes:
# leadr.common.domain.ids.EmailID.prefix¶
prefix = 'eml'
# leadr.common.domain.ids.EmailID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.GameID¶
Bases: PrefixedID
Game entity identifier.
Attributes:
# leadr.common.domain.ids.GameID.prefix¶
prefix = 'gam'
# leadr.common.domain.ids.GameID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.IdentityID¶
Bases: PrefixedID
Identity entity identifier.
Attributes:
# leadr.common.domain.ids.IdentityID.prefix¶
prefix = 'ide'
# leadr.common.domain.ids.IdentityID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.IdentitySessionID¶
Bases: PrefixedID
Identity session entity identifier.
Note: Uses same prefix as DeviceSessionID since it replaces that entity. During transition, both exist but reference different tables.
Attributes:
# leadr.common.domain.ids.IdentitySessionID.prefix¶
prefix = 'ses'
# leadr.common.domain.ids.IdentitySessionID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.JamCodeID¶
Bases: PrefixedID
Jam Code entity identifier.
Attributes:
# leadr.common.domain.ids.JamCodeID.prefix¶
prefix = 'jam'
# leadr.common.domain.ids.JamCodeID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.JamCodeRedemptionID¶
Bases: PrefixedID
Jam Code Redemption entity identifier.
Attributes:
# leadr.common.domain.ids.JamCodeRedemptionID.prefix¶
prefix = 'red'
# leadr.common.domain.ids.JamCodeRedemptionID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.NonceID¶
Bases: PrefixedID
Nonce entity identifier.
Attributes:
# leadr.common.domain.ids.NonceID.prefix¶
prefix = 'non'
# leadr.common.domain.ids.NonceID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.PrefixedID¶
PrefixedID(value=None)
Base class for entity IDs with type prefixes.
Provides Stripe-style prefixed UUIDs (e.g., "acc_123e4567-e89b-...") while maintaining internal UUID representation for database efficiency.
Usage
- AccountID() → generates new ID - AccountID("acc_123...") → parses prefixed string - AccountID(uuid_obj) → wraps existing UUIDAttributes:
Parameters:
- value (
str | UUID | Self | None) – Optional value to initialize from: - None: Generate new UUID
- str: Parse "prefix_uuid" format
- UUID: Wrap existing UUID
- PrefixedID: Extract UUID from another PrefixedID
Raises:
ValueError– If string format is invalid or prefix doesn't match
# leadr.common.domain.ids.PrefixedID.prefix¶
prefix: str = ''
# leadr.common.domain.ids.PrefixedID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.RunEntryID¶
Bases: PrefixedID
Run entry entity identifier.
Attributes:
# leadr.common.domain.ids.RunEntryID.prefix¶
prefix = 'run'
# leadr.common.domain.ids.RunEntryID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.ScoreEventID¶
Bases: PrefixedID
Score event entity identifier.
Attributes:
# leadr.common.domain.ids.ScoreEventID.prefix¶
prefix = 'sev'
# leadr.common.domain.ids.ScoreEventID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.ScoreFlagID¶
Bases: PrefixedID
Score flag entity identifier.
Attributes:
# leadr.common.domain.ids.ScoreFlagID.prefix¶
prefix = 'flg'
# leadr.common.domain.ids.ScoreFlagID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.ScoreID¶
Bases: PrefixedID
Score entity identifier.
Attributes:
# leadr.common.domain.ids.ScoreID.prefix¶
prefix = 'scr'
# leadr.common.domain.ids.ScoreID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.ScoreSubmissionMetaID¶
Bases: PrefixedID
Score submission metadata entity identifier.
Attributes:
# leadr.common.domain.ids.ScoreSubmissionMetaID.prefix¶
prefix = 'sub'
# leadr.common.domain.ids.ScoreSubmissionMetaID.uuid¶
uuid = uuid4()
leadr.common.domain.ids.UserID¶
Bases: PrefixedID
User entity identifier.
Attributes:
# leadr.common.domain.ids.UserID.prefix¶
prefix = 'usr'
# leadr.common.domain.ids.UserID.uuid¶
uuid = uuid4()
leadr.common.domain.models¶
Common domain models and value objects.
Classes:
- Entity – Base class for all domain entities with ID and timestamps.
- ImmutableEntity – Base class for immutable domain entities (append-only, no updates/deletes).
leadr.common.domain.models.Entity¶
Bases: BaseModel
Base class for all domain entities with ID and timestamps.
Provides common functionality for domain entities including:
- Auto-generated UUID primary key (or typed prefixed ID in subclasses)
- Created/updated timestamps (UTC)
- Soft delete support with deleted_at timestamp
- Equality and hashing based on ID
All domain entities should extend this base class. The ID and timestamps are automatically populated on entity creation and don't need to be provided by consumers.
Subclasses can override the id field with a typed PrefixedID for better
type safety and API clarity.
Functions:
- restore – Restore a soft-deleted entity.
- soft_delete – Mark entity as soft-deleted.
Attributes:
- created_at (
datetime) – - deleted_at (
datetime | None) – - id (
Any) – - is_deleted (
bool) – Check if entity is soft-deleted. - model_config –
- updated_at (
datetime) –
# leadr.common.domain.models.Entity.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
# leadr.common.domain.models.Entity.deleted_at¶
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
# leadr.common.domain.models.Entity.id¶
id: Any = Field(frozen=True, default_factory=uuid4, description='Unique identifier (auto-generated UUID or typed ID)')
# leadr.common.domain.models.Entity.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.common.domain.models.Entity.model_config¶
model_config = ConfigDict(validate_assignment=True)
# leadr.common.domain.models.Entity.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.common.domain.models.Entity.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.common.domain.models.Entity.updated_at¶
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
leadr.common.domain.models.ImmutableEntity¶
Bases: BaseModel
Base class for immutable domain entities (append-only, no updates/deletes).
Provides common functionality for event-sourced entities including:
- Auto-generated UUID primary key (or typed prefixed ID in subclasses)
- Created timestamp (UTC)
- Equality and hashing based on ID
Used for entities that are never updated or deleted after creation, such as ScoreEvent in event-sourcing patterns.
Subclasses can override the id field with a typed PrefixedID for better
type safety and API clarity.
Attributes:
- created_at (
datetime) – - id (
Any) – - model_config –
# leadr.common.domain.models.ImmutableEntity.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
# leadr.common.domain.models.ImmutableEntity.id¶
id: Any = Field(frozen=True, default_factory=uuid4, description='Unique identifier (auto-generated UUID or typed ID)')
# leadr.common.domain.models.ImmutableEntity.model_config¶
model_config = ConfigDict(validate_assignment=True)
leadr.common.domain.pagination¶
Pagination domain models.
Classes:
- CursorPosition – Position in a paginated result set.
- PaginationDirection – Direction of pagination.
- SortDirection – Sort direction for fields.
- SortField – Specification for sorting a field.
leadr.common.domain.pagination.CursorPosition¶
CursorPosition(values, entity_id)
Position in a paginated result set.
Attributes:
- values (
tuple[Any, ...]) – List of values for each sort field at this position - entity_id (
str) – The ID of the entity at this position (for stable sorting)
# leadr.common.domain.pagination.CursorPosition.entity_id¶
entity_id: str
# leadr.common.domain.pagination.CursorPosition.values¶
values: tuple[Any, ...]
leadr.common.domain.pagination.PaginationDirection¶
Direction of pagination.
Attributes:
# leadr.common.domain.pagination.PaginationDirection.BACKWARD¶
BACKWARD = 'backward'
# leadr.common.domain.pagination.PaginationDirection.FORWARD¶
FORWARD = 'forward'
leadr.common.domain.pagination.SortDirection¶
Sort direction for fields.
Attributes:
# leadr.common.domain.pagination.SortDirection.ASC¶
ASC = 'asc'
# leadr.common.domain.pagination.SortDirection.DESC¶
DESC = 'desc'
leadr.common.domain.pagination.SortField¶
SortField(name, direction)
Specification for sorting a field.
Attributes:
- name (
str) – Field name to sort by - direction (
SortDirection) – ASC or DESC
# leadr.common.domain.pagination.SortField.direction¶
direction: SortDirection
# leadr.common.domain.pagination.SortField.name¶
name: str
leadr.common.domain.pagination_result¶
Pagination result from repository layer.
Classes:
- PaginatedResult – Result of a paginated query from the repository layer.
Attributes:
- T –
leadr.common.domain.pagination_result.PaginatedResult¶
PaginatedResult(items, has_next, has_prev, next_position, prev_position)
Result of a paginated query from the repository layer.
Attributes:
- items (
list[T]) – List of entities returned by the query - has_next (
bool) – Whether there are more results after these items - has_prev (
bool) – Whether there are results before these items - next_position (
CursorPosition | None) – Position for the next page cursor (if has_next) - prev_position (
CursorPosition | None) – Position for the previous page cursor (if has_prev)
# leadr.common.domain.pagination_result.PaginatedResult.count¶
count: int
Return the number of items in this page.
# leadr.common.domain.pagination_result.PaginatedResult.has_next¶
has_next: bool
# leadr.common.domain.pagination_result.PaginatedResult.has_prev¶
has_prev: bool
# leadr.common.domain.pagination_result.PaginatedResult.items¶
items: list[T]
# leadr.common.domain.pagination_result.PaginatedResult.next_position¶
next_position: CursorPosition | None
# leadr.common.domain.pagination_result.PaginatedResult.prev_position¶
prev_position: CursorPosition | None
leadr.common.domain.pagination_result.T¶
T = TypeVar('T')
leadr.common.geoip¶
GeoIP service for IP address geolocation using MaxMind databases.
Classes:
- GeoIPService – Service for IP address geolocation using MaxMind GeoLite2 databases.
- GeoInfo – Geolocation information extracted from IP address.
Attributes:
- logger –
leadr.common.geoip.GeoIPService¶
GeoIPService(account_id, license_key, city_db_url, country_db_url, database_path, refresh_days=7, download_enabled=True)
Service for IP address geolocation using MaxMind GeoLite2 databases.
This service downloads and manages MaxMind GeoLite2 databases for IP geolocation. Databases are cached locally and refreshed periodically.
Example
> > > service = GeoIPService( > > > ... account_id="12345", > > > ... license_key="your_key", > > > ... city_db_url="https://download.maxmind.com/...", > > > ... country_db_url="https://download.maxmind.com/...", > > > ... database_path=Path(".geoip"), > > > ... refresh_days=7, > > > ... ) > > > await service.initialize() > > > geo_info = service.get_geo_info("8.8.8.8") > > > print(geo_info.country) > > > 'US'Functions:
- close – Close database readers and release resources.
- get_geo_info – Look up geolocation information for an IP address.
- initialize – Initialize the GeoIP service by downloading and loading databases.
Attributes:
- account_id –
- city_db_url –
- country_db_url –
- database_path –
- download_enabled –
- license_key –
- refresh_days –
Parameters:
- account_id (
str) – MaxMind account ID for basic auth - license_key (
str) – MaxMind license key for basic auth - city_db_url (
str) – URL to download GeoLite2 City database (tar.gz) - country_db_url (
str) – URL to download GeoLite2 Country database (tar.gz) - database_path (
Path) – Directory path to store database files - refresh_days (
int) – Number of days before refreshing databases (default: 7) - download_enabled (
bool) – Enable database downloads. Set to False when downloads are handled externally (e.g., init container). Default: True.
leadr.common.geoip.GeoIPService.account_id¶
account_id = account_id
leadr.common.geoip.GeoIPService.city_db_url¶
city_db_url = city_db_url
leadr.common.geoip.GeoIPService.close¶
close()
Close database readers and release resources.
leadr.common.geoip.GeoIPService.country_db_url¶
country_db_url = country_db_url
leadr.common.geoip.GeoIPService.database_path¶
database_path = database_path
leadr.common.geoip.GeoIPService.download_enabled¶
download_enabled = download_enabled
leadr.common.geoip.GeoIPService.get_geo_info¶
get_geo_info(ip_address)
Look up geolocation information for an IP address.
Parameters:
- ip_address (
str) – IP address to look up (e.g., '8.8.8.8')
Returns:
GeoInfo | None– GeoInfo with timezone, country, and city, or None if lookup fails
leadr.common.geoip.GeoIPService.initialize¶
initialize()
Initialize the GeoIP service by downloading and loading databases.
This method:
- Creates the database directory if it doesn't exist
- Downloads databases if they don't exist or are stale
- Extracts tar.gz files to get .mmdb files
- Opens database readers
Errors are logged but not raised - the service will work without databases (get_geo_info will return None).
leadr.common.geoip.GeoIPService.license_key¶
license_key = license_key
leadr.common.geoip.GeoIPService.refresh_days¶
refresh_days = refresh_days
leadr.common.geoip.GeoInfo¶
GeoInfo(timezone, country, city)
Geolocation information extracted from IP address.
Attributes:
- timezone (
str | None) – IANA timezone identifier (e.g., 'America/New_York') - country (
str | None) – ISO country code (e.g., 'US') - city (
str | None) – City name (e.g., 'New York')
leadr.common.geoip.GeoInfo.city¶
city: str | None
leadr.common.geoip.GeoInfo.country¶
country: str | None
leadr.common.geoip.GeoInfo.timezone¶
timezone: str | None
leadr.common.geoip.logger¶
logger = logging.getLogger(__name__)
leadr.common.orm¶
Common ORM base classes and utilities.
Classes:
- Base – Base class for all database models with UUID primary key and timestamps.
- ImmutableBase – Base class for immutable database models (append-only, no updates/deletes).
Attributes:
- nullable_timestamp –
- timestamp –
- uuid_pk –
leadr.common.orm.Base¶
Bases: DeclarativeBase
Base class for all database models with UUID primary key and timestamps.
Attributes:
- created_at (
Mapped[timestamp]) – - deleted_at (
Mapped[nullable_timestamp]) – - id (
Mapped[uuid_pk]) – - updated_at (
Mapped[timestamp]) –
leadr.common.orm.Base.created_at¶
created_at: Mapped[timestamp]
leadr.common.orm.Base.deleted_at¶
deleted_at: Mapped[nullable_timestamp]
leadr.common.orm.Base.id¶
id: Mapped[uuid_pk]
leadr.common.orm.Base.updated_at¶
updated_at: Mapped[timestamp] = mapped_column(onupdate=(func.now()))
leadr.common.orm.ImmutableBase¶
Bases: DeclarativeBase
Base class for immutable database models (append-only, no updates/deletes).
Used for event-sourcing entities like ScoreEvent that:
- Have no updated_at (immutable after creation)
- Have no deleted_at (append-only, never soft-deleted)
Attributes:
leadr.common.orm.ImmutableBase.created_at¶
created_at: Mapped[timestamp]
leadr.common.orm.ImmutableBase.id¶
id: Mapped[uuid_pk]
leadr.common.orm.ImmutableBase.metadata¶
metadata = Base.metadata
leadr.common.orm.ImmutableBase.registry¶
registry = Base.registry
leadr.common.orm.nullable_timestamp¶
nullable_timestamp = Annotated[datetime | None, mapped_column(DateTime(timezone=True), nullable=True, default=None)]
leadr.common.orm.timestamp¶
timestamp = Annotated[datetime, mapped_column(DateTime(timezone=True), nullable=False, server_default=(func.now()), default=(lambda: datetime.now(UTC)))]
leadr.common.orm.uuid_pk¶
uuid_pk = Annotated[UUID, mapped_column(primary_key=True, default=uuid4, server_default=(func.gen_random_uuid()))]
leadr.common.repositories¶
Base repository abstraction for common CRUD operations.
Classes:
- BaseRepository – Abstract base repository providing common CRUD operations.
- ImmutableBaseRepository – Abstract base repository for immutable (append-only) entities.
Attributes:
- DomainEntityT –
- ImmutableEntityT –
- ImmutableORMT –
- ORMModelT –
leadr.common.repositories.BaseRepository¶
BaseRepository(session)
Bases: ABC, Generic[DomainEntityT, ORMModelT]
Abstract base repository providing common CRUD operations.
All repositories should extend this class and implement the abstract methods for converting between domain entities and ORM models.
All delete operations are soft deletes by default, setting deleted_at timestamp.
Functions:
- create – Create a new entity in the database.
- delete – Soft delete an entity by setting its deleted_at timestamp.
- filter – Filter entities based on criteria with pagination.
- get_by_id – Get an entity by its ID.
- update – Update an existing entity in the database.
Attributes:
- session –
Parameters:
- session (
AsyncSession) – SQLAlchemy async session
leadr.common.repositories.BaseRepository.create¶
create(entity)
Create a new entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity to create
Returns:
DomainEntityT– Created domain entity with refreshed data
leadr.common.repositories.BaseRepository.delete¶
delete(entity_id)
Soft delete an entity by setting its deleted_at timestamp.
Parameters:
- entity_id (
UUID4 | PrefixedID) – ID of entity to delete
Raises:
EntityNotFoundError– If entity is not found
leadr.common.repositories.BaseRepository.filter¶
filter(account_id=None, *, pagination, **kwargs)
Filter entities based on criteria with pagination.
All filter operations return paginated results. The pagination parameter is required to enforce consistent API behavior across the codebase.
For multi-tenant entities, implementations should make account_id required (no default). For top-level entities like Account, account_id can remain optional and unused.
Parameters:
- account_id (
AccountID | None) – Optional account ID for filtering. Multi-tenant entities should override to make this required. - pagination (
PaginationParams) – Required pagination parameters (cursor, limit, sort). - **kwargs (
Any) – Additional filter parameters specific to the entity type.
Returns:
PaginatedResult[DomainEntityT]– PaginatedResult containing matching entities and pagination metadata.
leadr.common.repositories.BaseRepository.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.common.repositories.BaseRepository.session¶
session = session
leadr.common.repositories.BaseRepository.update¶
update(entity)
Update an existing entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity with updated data
Returns:
DomainEntityT– Updated domain entity with refreshed data
Raises:
EntityNotFoundError– If entity is not found
leadr.common.repositories.DomainEntityT¶
DomainEntityT = TypeVar('DomainEntityT', bound=Entity)
leadr.common.repositories.ImmutableBaseRepository¶
ImmutableBaseRepository(session)
Bases: ABC, Generic[ImmutableEntityT, ImmutableORMT]
Abstract base repository for immutable (append-only) entities.
Used for event-sourced entities that are never updated or deleted. Provides only create, get, and filter operations.
Functions:
- create – Create a new immutable entity in the database.
- filter – Filter immutable entities based on criteria with pagination.
- get_by_id – Get an immutable entity by its ID.
Attributes:
- session –
Parameters:
- session (
AsyncSession) – SQLAlchemy async session
leadr.common.repositories.ImmutableBaseRepository.create¶
create(entity)
Create a new immutable entity in the database.
Parameters:
- entity (
ImmutableEntityT) – Domain entity to create
Returns:
ImmutableEntityT– Created domain entity with refreshed data
leadr.common.repositories.ImmutableBaseRepository.filter¶
filter(account_id=None, *, pagination, **kwargs)
Filter immutable entities based on criteria with pagination.
Parameters:
- account_id (
AccountID | None) – Optional account ID for filtering. - pagination (
PaginationParams) – Required pagination parameters (cursor, limit, sort). - **kwargs (
Any) – Additional filter parameters specific to the entity type.
Returns:
PaginatedResult[ImmutableEntityT]– PaginatedResult containing matching entities and pagination metadata.
leadr.common.repositories.ImmutableBaseRepository.get_by_id¶
get_by_id(entity_id)
Get an immutable entity by its ID.
Parameters:
- entity_id (
UUID4 | PrefixedID) – Entity ID to retrieve
Returns:
ImmutableEntityT | None– Domain entity if found, None otherwise
leadr.common.repositories.ImmutableBaseRepository.session¶
session = session
leadr.common.repositories.ImmutableEntityT¶
ImmutableEntityT = TypeVar('ImmutableEntityT', bound=ImmutableEntity)
leadr.common.repositories.ImmutableORMT¶
ImmutableORMT = TypeVar('ImmutableORMT', bound=ImmutableBase)
leadr.common.repositories.ORMModelT¶
ORMModelT = TypeVar('ORMModelT', bound=Base)
leadr.common.services¶
Base service abstraction for common business logic patterns.
Classes:
- BaseService – Abstract base service providing common business logic patterns.
Attributes:
- DomainEntityT –
- RepositoryT –
leadr.common.services.BaseService¶
BaseService(session, repository=None)
Bases: ABC, Generic[DomainEntityT, RepositoryT]
Abstract base service providing common business logic patterns.
All services should extend this class to ensure consistent patterns for:
- Repository initialization
- Standard CRUD operations
- Error handling with domain exceptions
The service layer sits between API routes and repositories, providing:
- Business logic orchestration
- Domain validation
- Transaction boundaries
- Consistent error handling
Functions:
- 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.
- list_all – List all non-deleted entities.
- soft_delete – Soft-delete an entity and return it before deletion.
Attributes:
- repository –
Parameters:
- session (
AsyncSession) – SQLAlchemy async session for database operations - repository (
RepositoryT | None) – Optional pre-built repository instance. If provided, skips _create_repository(). Useful for injecting mock repositories in tests.
leadr.common.services.BaseService.delete¶
delete(entity_id)
Soft-delete an entity.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to delete
Raises:
EntityNotFoundError– If the entity doesn't exist
leadr.common.services.BaseService.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.common.services.BaseService.get_by_id_or_raise¶
get_by_id_or_raise(entity_id)
Get an entity by its ID or raise EntityNotFoundError.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to retrieve
Returns:
DomainEntityT– The domain entity
Raises:
EntityNotFoundError– If the entity is not found (converted to HTTP 404 by global handler)
leadr.common.services.BaseService.list_all¶
list_all()
List all non-deleted entities.
Returns:
list[DomainEntityT]– List of domain entities
leadr.common.services.BaseService.repository¶
repository = repository if repository is not None else self._create_repository(session)
leadr.common.services.BaseService.soft_delete¶
soft_delete(entity_id)
Soft-delete an entity and return it before deletion.
Useful for endpoints that need to return the deleted entity in the response.
Parameters:
- entity_id (
UUID | PrefixedID) – The ID of the entity to delete
Returns:
DomainEntityT– The entity before it was deleted
Raises:
EntityNotFoundError– If the entity doesn't exist
leadr.common.services.DomainEntityT¶
DomainEntityT = TypeVar('DomainEntityT', bound=Entity)
leadr.common.services.RepositoryT¶
RepositoryT = TypeVar('RepositoryT', bound=BaseRepository)
leadr.common.utils¶
Common utility functions.
Modules:
leadr.common.utils.ip¶
IP address extraction utilities.
Functions:
- extract_client_ip – Extract client IP address from request headers.
leadr.common.utils.ip.extract_client_ip¶
extract_client_ip(request)
Extract client IP address from request headers.
Checks headers in priority order:
- X-Real-IP - common proxy header
- X-Forwarded-For - standard proxy header (uses leftmost IP)
- CF-Connecting-IP - Cloudflare header
- request.client.host - fallback to direct connection
Parameters:
- request (
Request) – The incoming FastAPI/Starlette request
Returns:
str | None– IP address string, or None if unable to extract
leadr.common.utils.slug¶
Slug generation utilities.
Provides utilities for generating URL-friendly slugs from text, with support for collision handling and uniqueness constraints.
Functions:
- generate_slug – Generate a URL-friendly slug from text.
- generate_unique_slug_with_retry – Generate a unique slug with collision handling.
Attributes:
- T –
leadr.common.utils.slug.T¶
T = TypeVar('T')
leadr.common.utils.slug.generate_slug¶
generate_slug(text, max_length=50)
Generate a URL-friendly slug from text.
Converts text to lowercase, replaces spaces with hyphens, removes special characters, and handles unicode by transliterating accented characters.
Parameters:
- text (
str) – The text to convert to a slug - max_length (
int) – Maximum length of the generated slug (default: 50)
Returns:
str– A lowercase slug with alphanumeric characters, hyphens, and underscores
Raises:
ValueError– If text is empty or contains no valid characters after processing
Examples:
>>> generate_slug("Hello World")
"hello-world"
>>> generate_slug("Café Board")
"cafe-board"
>>> generate_slug("Speed-Run 2024")
"speed-run-2024"
leadr.common.utils.slug.generate_unique_slug_with_retry¶
generate_unique_slug_with_retry(base_text, check_exists, max_retries=10, max_length=50)
Generate a unique slug with collision handling.
Attempts to generate a unique slug from the base text. If a collision is detected (slug already exists), appends a numeric suffix and retries.
Parameters:
- base_text (
str) – The text to convert to a slug - check_exists (
Callable[[str], Awaitable[bool]]) – Async function that returns True if slug already exists - max_retries (
int) – Maximum number of collision retries (default: 10) - max_length (
int) – Maximum length of the generated slug (default: 50)
Returns:
str– A unique slug that doesn't collide with existing slugs
Raises:
ValueError– If unable to generate unique slug after max_retriesValueError– If base_text is invalid for slug generation
Examples:
>>> async def check(slug: str) -> bool:
... return slug in ["my-board", "my-board-2"]
>>> await generate_unique_slug_with_retry("My Board", check)
"my-board-3"