deps¶

Documentation for the deps module.

Application dependency providers generators.

This module contains functions to create dependency providers for services and filters.

You should not have modify this module very often and should only be invoked under normal usage.

class app.lib.deps.CompositeServiceMixin[source]¶

Bases: object

Mixin for services that orchestrate multiple repositories.

Provides lazy instantiation of dependent services that share the parent service’s database session.

Example

```python from app.lib.deps import CompositeServiceMixin

class UserService(CompositeServiceMixin, SQLAlchemyAsyncRepositoryService[m.User]):

@property def oauth_accounts(self) -> UserOAuthAccountService:

return self._get_service(UserOAuthAccountService)

async def authenticate_oauth_user(self, …) -> m.User:: await self.oauth_accounts.create_or_update_oauth_account(…)

```

app.lib.deps.create_filter_dependencies(config: ~advanced_alchemy.extensions.litestar.providers.FilterConfig, dep_defaults: ~advanced_alchemy.extensions.litestar.providers.DependencyDefaults = <advanced_alchemy.extensions.litestar.providers.DependencyDefaults object>) → dict[str, Provide][source]¶

Create a dependency provider for the combined filter function.

Parameters:

config – FilterConfig instance with desired settings.
dep_defaults – Dependency defaults to use for the filter dependencies

Returns:

A dependency provider function for the combined filter function.

app.lib.deps.create_service_dependencies(service_class: type[AsyncServiceT_co | SyncServiceT_co], /, key: str, statement: Optional[Select[tuple[ModelT]]] = None, config: Optional[Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]] = None, error_messages: Optional[Union[ErrorMessages, EmptyType]] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: Optional[LoadSpec] = None, execution_options: Optional[dict[str, Any]] = None, filters: Optional[FilterConfig] = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, dep_defaults: DependencyDefaults = <advanced_alchemy.extensions.litestar.providers.DependencyDefaults object>) → dict[str, Provide][source]¶

Create a dependency provider for the combined filter function.

Parameters:

key – The key to use for the dependency provider.
service_class – The service class to create a dependency provider for.
statement – The statement to use for the service.
config – The configuration to use for the service.
error_messages – The error messages to use for the service.
load – The load spec to use for the service.
execution_options – The execution options to use for the service.
filters – The filter configuration to use for the service.
uniquify – Whether to uniquify the service.
count_with_window_function – Whether to count with a window function.
dep_defaults – The dependency defaults to use for the service.

Returns:

A dictionary of dependency providers for the service.

app.lib.deps.create_service_provider(service_class: type[AsyncServiceT_co | SyncServiceT_co], /, statement: Optional[Select[tuple[ModelT]]] = None, config: Optional[Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]] = None, error_messages: Optional[Union[ErrorMessages, EmptyType]] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: Optional[LoadSpec] = None, execution_options: Optional[dict[str, Any]] = None, uniquify: bool | None = None, count_with_window_function: bool | None = None) → Callable[[...], AsyncGenerator[AsyncServiceT_co, None] | Generator[SyncServiceT_co, None, None]][source]¶

Create a dependency provider for a service with a configurable session key.

Parameters:

service_class – The service class inheriting from SQLAlchemyAsyncRepositoryService or SQLAlchemySyncRepositoryService.
statement – An optional SQLAlchemy Select statement to scope the service.
config – An optional SQLAlchemy configuration object.
error_messages – Optional custom error messages for the service.
load – Optional LoadSpec for eager loading relationships.
execution_options – Optional dictionary of execution options for SQLAlchemy.
uniquify – Optional flag to uniquify results.
count_with_window_function – Optional flag to use window function for counting.

Returns:

A dependency provider function suitable for Litestar’s DI system.

async app.lib.deps.get_task_queue() → Queue[source]¶

Get the background task queue.

Returns:: The connected background task queue.
Return type:: Queue

app.lib.deps.provide_services(p1: ServiceProvider[T1], /, *, session: AsyncSession | None = None, connection: ASGIConnection[Any, Any, Any, Any] | None = None) → AbstractAsyncContextManager[tuple[T1]][source]¶

app.lib.deps.provide_services(p1: ServiceProvider[T1], p2: ServiceProvider[T2], /, *, session: AsyncSession | None = None, connection: ASGIConnection[Any, Any, Any, Any] | None = None) → AbstractAsyncContextManager[tuple[T1, T2]]

app.lib.deps.provide_services(p1: ServiceProvider[T1], p2: ServiceProvider[T2], p3: ServiceProvider[T3], /, *, session: AsyncSession | None = None, connection: ASGIConnection[Any, Any, Any, Any] | None = None) → AbstractAsyncContextManager[tuple[T1, T2, T3]]

app.lib.deps.provide_services(p1: ServiceProvider[T1], p2: ServiceProvider[T2], p3: ServiceProvider[T3], p4: ServiceProvider[T4], /, *, session: AsyncSession | None = None, connection: ASGIConnection[Any, Any, Any, Any] | None = None) → AbstractAsyncContextManager[tuple[T1, T2, T3, T4]]

app.lib.deps.provide_services(p1: ServiceProvider[T1], p2: ServiceProvider[T2], p3: ServiceProvider[T3], p4: ServiceProvider[T4], p5: ServiceProvider[T5], /, *, session: AsyncSession | None = None, connection: ASGIConnection[Any, Any, Any, Any] | None = None) → AbstractAsyncContextManager[tuple[T1, T2, T3, T4, T5]]

Provide multiple services sharing the same database session.

This context manager simplifies acquiring services outside of Litestar’s dependency injection context (background jobs, CLI commands, event handlers, and authentication guards).

Parameters:

*providers – One or more service provider functions created via create_service_provider(). Each provider should accept an AsyncSession and yield a service instance.
session – Optional pre-existing database session. If provided, the session lifecycle is NOT managed by this context manager. The caller is responsible for session cleanup.
connection – Optional ASGI connection for request-scoped contexts. Used in authentication guards to obtain the session from the request scope via alchemy.provide_session().

Yields:

A tuple of instantiated services, matching the order of providers passed as arguments.

Raises:

ValueError – If both session and connection are provided, or if no providers are given.

Examples

Standalone context (jobs, CLI, events) - auto-creates session:

async with provide_services(
    provide_email_verification_service,
    provide_password_reset_service,
) as (verification_service, reset_service):
    await verification_service.cleanup_expired_tokens()
    await reset_service.cleanup_expired_tokens()

Request context (guards) - uses connection-scoped session:

async with provide_services(
    provide_users_service,
    connection=connection,
) as (users_service,):
    user = await users_service.get_one_or_none(email=token.sub)

Explicit session - caller manages lifecycle:

async with alchemy.get_session() as db_session:
    async with provide_services(
        provide_users_service,
        session=db_session,
    ) as (users_service,):
        await users_service.create(data=user_data)
    # Additional operations with db_session...