Feature: CompositeService base class for multi-repository orchestration

[cofin]

Summary

When business logic requires coordinating multiple repositories/models, developers need to instantiate multiple services that share the same database session. Currently this is done ad-hoc within service methods, leading to hidden dependencies and boilerplate.

Current Pattern (Ad-hoc)

class UserService(SQLAlchemyAsyncRepositoryService[User]):
    async def authenticate_or_create_oauth_user(self, provider: str, oauth_data: dict) -> User:
        # Hidden dependency - created inline
        from app.domain.accounts.services import UserOAuthAccountService
        
        existing_user = await self.get_one_or_none(email=oauth_data['email'])
        
        # Instantiate sibling service with shared session
        oauth_service = UserOAuthAccountService(session=self.repository.session)
        await oauth_service.create_or_update_oauth_account(...)
        
        return existing_user

class TeamService(SQLAlchemyAsyncRepositoryService[Team]):
    async def _populate_with_tags(self, data: dict) -> dict:
        # Another hidden dependency
        from app.domain.tags.services import TagService
        
        tag_service = TagService(session=self.repository.session)
        data.tags.extend([await tag_service.upsert({'name': t}) for t in tags])
        return data

Issues:

• Hidden dependencies (imports inside methods)
• Repeated Service(session=self.repository.session) boilerplate
• Hard to test (can't easily mock inner services)
• No clear "orchestration" layer for complex workflows

Proposed: CompositeService Base Class

A base class that composes multiple services with a shared session:

from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService, CompositeService

class UserRegistrationService(CompositeService):
    """Orchestrates user registration workflow across multiple services."""
    
    # Declare dependent services as class attributes with type annotations
    users: UserService
    roles: RoleService
    verification: EmailVerificationTokenService
    audit: AuditLogService
    
    async def register(self, data: UserCreate) -> User:
        # All services auto-instantiated with shared session
        user = await self.users.create(data.to_dict())
        
        # Assign default role
        default_role = await self.roles.get_one_or_none(slug='user')
        if default_role:
            user.roles.append(UserRole(role_id=default_role.id))
            await self.users.update(item_id=user.id, data=user)
        
        # Create verification token
        await self.verification.create_token(user_id=user.id)
        
        # Audit log
        await self.audit.log_action('user.registered', target_id=user.id)
        
        return user
    
    async def register_oauth(self, provider: str, oauth_data: dict) -> tuple[User, bool]:
        """Register or link OAuth account."""
        existing = await self.users.get_one_or_none(email=oauth_data['email'])
        if existing:
            # ... link account logic
            return existing, False
        
        user = await self.register(UserCreate.from_oauth(oauth_data))
        return user, True

Implementation Sketch

from typing import Any, get_type_hints

class CompositeService:
    """Base class for services that orchestrate multiple repositories."""
    
    def __init__(
        self,
        session: AsyncSession | None = None,
        config: SQLAlchemyAsyncConfig | None = None,
    ) -> None:
        if not session and not config:
            raise ValueError('Must provide session or config')
        
        self._session = session
        self._config = config
        self._services: dict[str, Any] = {}
        
        # Auto-instantiate declared services from class annotations
        hints = get_type_hints(self.__class__)
        for name, service_cls in hints.items():
            if isinstance(service_cls, type) and issubclass(service_cls, SQLAlchemyAsyncRepositoryService):
                # Lazy instantiation or eager - TBD
                pass
    
    def __getattr__(self, name: str) -> Any:
        """Lazy-instantiate services on first access."""
        hints = get_type_hints(self.__class__)
        if name in hints:
            service_cls = hints[name]
            if name not in self._services:
                self._services[name] = service_cls(session=self._session)
            return self._services[name]
        raise AttributeError(f'{self.__class__.__name__} has no attribute {name}')
    
    @classmethod
    @asynccontextmanager
    async def new(
        cls,
        session: AsyncSession | None = None,
        config: SQLAlchemyAsyncConfig | None = None,
    ) -> AsyncIterator[Self]:
        """Context manager for creating composite service."""
        if session:
            yield cls(session=session)
        elif config:
            async with config.get_session() as db_session:
                yield cls(session=db_session)
        else:
            raise ValueError('Must provide session or config')

Usage with Dependency Injection

Works seamlessly with existing patterns:

# Create provider
provide_user_registration = create_service_provider(UserRegistrationService)

# In controller
class AccessController(Controller):
    dependencies = {
        'registration_service': Provide(provide_user_registration),
    }
    
    @post('/signup')
    async def signup(
        self, 
        registration_service: UserRegistrationService,
        data: UserCreate,
    ) -> User:
        return await registration_service.register(data)

Usage Outside DI (with proposed provide_services from #652)

# In background job
async with provide_services(provide_user_registration) as (registration,):
    await registration.cleanup_unverified_users()

# Or with Service.new()
async with UserRegistrationService.new(config=alchemy) as registration:
    await registration.register(data)

Benefits

1. Explicit dependencies: Class attributes declare what's composed
2. Shared session: All services use same session/transaction boundary
3. Type-safe: IDE autocompletion for all composed services
4. Testable: Easy to mock individual services via constructor
5. Lazy instantiation: Services created only when accessed
6. Works with existing patterns: Providers, DI, .new() context manager

Alternative Names

• CompositeService
• MultiService
• OrchestratorService
• AggregateService
• ServiceComposite

Related

• Feature: Add provide_services context manager for multi-service acquisition outside DI #652 - provide_services context manager for multi-service acquisition
• Unit of Work pattern (this is a lighter-weight alternative)

Would love feedback on this approach! Happy to contribute a PR.