Infrastructure Utilities

Infrastructure-level utilities for service orchestration and caching strategies.

Service Orchestration

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

class ExerciseService(CompositeServiceMixin, SQLAlchemyAsyncRepositoryService[m.Exercise]):
    @property
    def muscles(self) -> MuscleGroupService:
        return self._get_service(MuscleGroupService)

    async def process_exercise(self, exercise_id: UUID) -> None:
        # Accessing the cached service instance
        muscle_groups = await self.muscles.get_all()

Cache Infrastructure

class CacheKeyBuilder[source]

Bases: object

Utility for generating cache keys for complex domain-specific scenarios.

classmethod for_exercises(params: ExerciseFilters, user_id: UUID) str[source]

Extend the filter cache key with a user ID for data isolation.

This ensures that cached results for private or shared scopes are correctly partitioned between different users.