Source code for app.domain.teams.dependencies

"""User Account Controllers."""

from __future__ import annotations

from typing import TYPE_CHECKING

from sqlalchemy.orm import joinedload, noload, selectinload

from app.db.models import Team, TeamInvitation, TeamMember
from app.domain.teams.services import TeamInvitationService, TeamMemberService, TeamService

__all__ = ("provide_team_members_service", "provide_teams_service", "provide_team_invitations_service")


if TYPE_CHECKING:
    from collections.abc import AsyncGenerator

    from sqlalchemy.ext.asyncio import AsyncSession


[docs] async def provide_teams_service(db_session: AsyncSession) -> AsyncGenerator[TeamService, None]: """Construct repository and service objects for the request.""" async with TeamService.new( session=db_session, load=[ selectinload(Team.tags), selectinload(Team.members).options( joinedload(TeamMember.user, innerjoin=True), ), ], ) as service: yield service
[docs] async def provide_team_members_service(db_session: AsyncSession) -> AsyncGenerator[TeamMemberService, None]: """Construct repository and service objects for the request.""" async with TeamMemberService.new( session=db_session, load=[ noload("*"), joinedload(TeamMember.team, innerjoin=True).options(noload("*")), joinedload(TeamMember.user, innerjoin=True).options(noload("*")), ], ) as service: yield service
[docs] async def provide_team_invitations_service(db_session: AsyncSession) -> AsyncGenerator[TeamInvitationService, None]: """Construct repository and service objects for the request.""" async with TeamInvitationService.new( session=db_session, load=[ noload("*"), joinedload(TeamInvitation.team, innerjoin=True).options(noload("*")), joinedload(TeamInvitation.invited_by, innerjoin=True).options(noload("*")), ], ) as service: yield service