Skip to content

Feature: Add provide_services context manager for multi-service acquisition outside DI #652

@cofin

Description

@cofin

Summary

When using Advanced Alchemy services outside of Litestar's dependency injection context (background jobs, CLI commands, event handlers, authentication guards), developers must manually manage session acquisition and consume service providers with anext(). This leads to repetitive boilerplate code.

Current Pattern (Verbose)

from app.config import alchemy
from app.domain.accounts.dependencies import (
    provide_email_verification_service,
    provide_password_reset_service,
    provide_refresh_token_service,
)

async def cleanup_auth_tokens() -> dict[str, int]:
    async with alchemy.get_session() as db_session:
        verification_service = await anext(provide_email_verification_service(db_session))
        reset_service = await anext(provide_password_reset_service(db_session))
        refresh_service = await anext(provide_refresh_token_service(db_session))

        verification_count = await verification_service.cleanup_expired_tokens()
        reset_count = await reset_service.cleanup_expired_tokens()
        refresh_count = await refresh_service.cleanup_expired_tokens()
    
    return {...}

Proposed API

A provide_services context manager that simplifies this pattern:

from advanced_alchemy.extensions.litestar.providers import provide_services

async def cleanup_auth_tokens() -> dict[str, int]:
    async with provide_services(
        provide_email_verification_service,
        provide_password_reset_service,
        provide_refresh_token_service,
        config=alchemy,  # Auto-creates session
    ) as (verification_service, reset_service, refresh_service):
        verification_count = await verification_service.cleanup_expired_tokens()
        reset_count = await reset_service.cleanup_expired_tokens()
        refresh_count = await refresh_service.cleanup_expired_tokens()
    
    return {...}

Proposed Features

  1. Tuple unpacking: Returns services as a tuple matching provider order
  2. Session sources:
    • config=: Auto-creates and manages session lifecycle (standalone contexts)
    • session=: Uses provided session (caller manages lifecycle)
    • connection=: Gets session from ASGI connection scope (request contexts)
  3. Type safety: Overloads for 1-5 providers with proper tuple return types
  4. Validation: Raises ValueError for invalid parameter combinations

Use Cases

  • Background jobs (SAQ, Celery, etc.): Token cleanup, email sending, data sync
  • CLI commands: User management, database seeding, maintenance
  • Event handlers: Post-signup flows, notification triggers
  • Auth guards: JWT validation requiring user lookup

Implementation Sketch

from contextlib import asynccontextmanager
from typing import overload, TypeVar, Any

@overload
@asynccontextmanager
async def provide_services(
    __p1: ServiceProvider[S1],
    /,
    *,
    config: SQLAlchemyAsyncConfig | None = ...,
    session: AsyncSession | None = ...,
) -> AsyncGenerator[tuple[S1], None]: ...

@overload
@asynccontextmanager
async def provide_services(
    __p1: ServiceProvider[S1],
    __p2: ServiceProvider[S2],
    /,
    *,
    config: SQLAlchemyAsyncConfig | None = ...,
    session: AsyncSession | None = ...,
) -> AsyncGenerator[tuple[S1, S2], None]: ...

# ... overloads for 3, 4, 5 providers

@asynccontextmanager
async def provide_services(
    *providers,
    config=None,
    session=None,
):
    if session is not None:
        services = tuple([await anext(p(session)) for p in providers])
        yield services
    elif config is not None:
        async with config.get_session() as db_session:
            services = tuple([await anext(p(db_session)) for p in providers])
            yield services
    else:
        raise ValueError("Must provide either 'config' or 'session'")

Benefits

  • ~70% boilerplate reduction for multi-service operations
  • Consistent pattern across all out-of-DI contexts
  • Type-safe with IDE autocompletion
  • Session lifecycle clarity - owned vs borrowed semantics

Related

This pattern complements the existing create_service_provider factory and Service.new() context manager.

Would love to hear thoughts on this! Happy to contribute a PR if there's interest.

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions