jacobh2
ac3b6a41d7
Deleting a period left every day it had funded behind. The link from a day back to its period was set to null the instant the period was deleted, so the days became untraceable before anything could clean them up, and the next recalculation had no way to know they no longer belonged to anything. Undo had the same fault for the same reason: the period went, the days stayed, and a reload showed them still there. The days are now removed with the period, before the link can be broken. A day pinned by hand goes too — pinning keeps the planner from quietly retyping a day, not from being deleted when you delete the period it belongs to, which is already how an unplanned band behaves.
380 lines
14 KiB
Python
380 lines
14 KiB
Python
"""Dependency-injection wiring.
|
|
|
|
Each factory picks the correct concrete repo based on `settings.database_type`.
|
|
Routers/services depend only on the abstract interfaces.
|
|
|
|
Pattern (copy for each new entity):
|
|
1. `get_<entity>_repository` — selects sqlite vs postgres impl
|
|
2. `<Entity>RepositoryDep` — Annotated type alias
|
|
3. `get_<entity>_service` — wires repo(s) into the service
|
|
4. `<Entity>ServiceDep` — Annotated type alias
|
|
"""
|
|
|
|
from typing import Annotated
|
|
|
|
import aiosqlite
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from app.config import settings
|
|
from app.entities.user import User
|
|
from app.repositories.child_repository import ChildRepository
|
|
from app.repositories.day_override_repository import DayOverrideRepository
|
|
from app.repositories.leave_wish_repository import LeaveWishRepository
|
|
from app.repositories.parental_leave_repository import ParentalLeaveRepository
|
|
from app.repositories.personal_data_repository import PersonalDataRepository
|
|
from app.repositories.planned_leave_repository import PlannedLeaveRepository
|
|
from app.repositories.tip_dismissal_repository import TipDismissalRepository
|
|
from app.repositories.user_repository import UserRepository
|
|
from app.repositories.vacation_days_repository import VacationDaysRepository
|
|
from app.repositories.workspace_repository import (
|
|
WorkspaceInvitationRepository,
|
|
WorkspaceRepository,
|
|
)
|
|
from app.repositories_impl.sqlite.child_repository import SQLiteChildRepository
|
|
from app.repositories_impl.sqlite.day_override_repository import SQLiteDayOverrideRepository
|
|
from app.repositories_impl.sqlite.leave_wish_repository import SQLiteLeaveWishRepository
|
|
from app.repositories_impl.sqlite.parental_leave_repository import SQLiteParentalLeaveRepository
|
|
from app.repositories_impl.sqlite.personal_data_repository import SQLitePersonalDataRepository
|
|
from app.repositories_impl.sqlite.planned_leave_repository import SQLitePlannedLeaveRepository
|
|
from app.repositories_impl.sqlite.tip_dismissal_repository import SQLiteTipDismissalRepository
|
|
from app.repositories_impl.sqlite.user_repository import SQLiteUserRepository
|
|
from app.repositories_impl.sqlite.vacation_days_repository import SQLiteVacationDaysRepository
|
|
from app.repositories_impl.sqlite.workspace_repository import (
|
|
SQLiteWorkspaceInvitationRepository,
|
|
SQLiteWorkspaceRepository,
|
|
)
|
|
from app.services.allocation_service import AllocationInputLoader, PlanningEngineService
|
|
from app.services.analytics_service import AnalyticsService
|
|
from app.services.auth_service import AuthService
|
|
from app.services.calendar_service import CalendarService
|
|
from app.services.child_service import ChildService
|
|
from app.services.day_override_service import DayOverrideService
|
|
from app.services.leave_wish_service import LeaveWishService
|
|
from app.services.personal_data_service import PersonalDataService
|
|
from app.services.planning_service import PlanningService
|
|
from app.services.tips_service import TipsService
|
|
from app.services.vacation_days_service import VacationDaysService
|
|
from app.services.workspace_service import WorkspaceService
|
|
from app.utils.errors import UnauthorizedError
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DB connection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_connection(request: Request) -> aiosqlite.Connection:
|
|
"""Return the process-wide DB connection opened during app startup."""
|
|
return request.app.state.db_connection
|
|
|
|
|
|
ConnectionDep = Annotated[aiosqlite.Connection, Depends(get_connection)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Repository factories
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _sqlite_or_error(label: str) -> None:
|
|
if settings.database_type not in ("sqlite", "postgres"):
|
|
raise ValueError(f"Unsupported database_type for {label}: {settings.database_type}")
|
|
|
|
|
|
def get_user_repository(conn: ConnectionDep) -> UserRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteUserRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
UserRepositoryDep = Annotated[UserRepository, Depends(get_user_repository)]
|
|
|
|
|
|
def get_personal_data_repository(conn: ConnectionDep) -> PersonalDataRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLitePersonalDataRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
PersonalDataRepositoryDep = Annotated[PersonalDataRepository, Depends(get_personal_data_repository)]
|
|
|
|
|
|
def get_workspace_repository(conn: ConnectionDep) -> WorkspaceRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteWorkspaceRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
WorkspaceRepositoryDep = Annotated[WorkspaceRepository, Depends(get_workspace_repository)]
|
|
|
|
|
|
def get_workspace_invitation_repository(conn: ConnectionDep) -> WorkspaceInvitationRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteWorkspaceInvitationRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
WorkspaceInvitationRepositoryDep = Annotated[
|
|
WorkspaceInvitationRepository, Depends(get_workspace_invitation_repository)
|
|
]
|
|
|
|
|
|
def get_child_repository(conn: ConnectionDep) -> ChildRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteChildRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
ChildRepositoryDep = Annotated[ChildRepository, Depends(get_child_repository)]
|
|
|
|
|
|
def get_parental_leave_repository(conn: ConnectionDep) -> ParentalLeaveRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteParentalLeaveRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
ParentalLeaveRepositoryDep = Annotated[
|
|
ParentalLeaveRepository, Depends(get_parental_leave_repository)
|
|
]
|
|
|
|
|
|
def get_planned_leave_repository(conn: ConnectionDep) -> PlannedLeaveRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLitePlannedLeaveRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
PlannedLeaveRepositoryDep = Annotated[PlannedLeaveRepository, Depends(get_planned_leave_repository)]
|
|
|
|
|
|
def get_vacation_days_repository(conn: ConnectionDep) -> VacationDaysRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteVacationDaysRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
VacationDaysRepositoryDep = Annotated[VacationDaysRepository, Depends(get_vacation_days_repository)]
|
|
|
|
|
|
def get_leave_wish_repository(conn: ConnectionDep) -> LeaveWishRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteLeaveWishRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
LeaveWishRepositoryDep = Annotated[LeaveWishRepository, Depends(get_leave_wish_repository)]
|
|
|
|
|
|
def get_day_override_repository(conn: ConnectionDep) -> DayOverrideRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteDayOverrideRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
DayOverrideRepositoryDep = Annotated[DayOverrideRepository, Depends(get_day_override_repository)]
|
|
|
|
|
|
def get_tip_dismissal_repository(conn: ConnectionDep) -> TipDismissalRepository:
|
|
if settings.database_type == "sqlite":
|
|
return SQLiteTipDismissalRepository(conn)
|
|
raise ValueError(f"Unsupported database_type: {settings.database_type}")
|
|
|
|
|
|
TipDismissalRepositoryDep = Annotated[TipDismissalRepository, Depends(get_tip_dismissal_repository)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Service factories
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_auth_service(
|
|
user_repo: UserRepositoryDep,
|
|
pd_repo: PersonalDataRepositoryDep,
|
|
ws_repo: WorkspaceRepositoryDep,
|
|
) -> AuthService:
|
|
return AuthService(user_repo, pd_repo, ws_repo)
|
|
|
|
|
|
AuthServiceDep = Annotated[AuthService, Depends(get_auth_service)]
|
|
|
|
|
|
def get_vacation_days_service(repo: VacationDaysRepositoryDep) -> VacationDaysService:
|
|
return VacationDaysService(repo)
|
|
|
|
|
|
VacationDaysServiceDep = Annotated[VacationDaysService, Depends(get_vacation_days_service)]
|
|
|
|
|
|
def get_leave_wish_service(
|
|
repo: LeaveWishRepositoryDep,
|
|
planned_leave_repo: PlannedLeaveRepositoryDep,
|
|
) -> LeaveWishService:
|
|
return LeaveWishService(repo, planned_leave_repo)
|
|
|
|
|
|
LeaveWishServiceDep = Annotated[LeaveWishService, Depends(get_leave_wish_service)]
|
|
|
|
|
|
def get_day_override_service(repo: DayOverrideRepositoryDep) -> DayOverrideService:
|
|
return DayOverrideService(repo)
|
|
|
|
|
|
DayOverrideServiceDep = Annotated[DayOverrideService, Depends(get_day_override_service)]
|
|
|
|
|
|
def get_calendar_service(day_override_service: DayOverrideServiceDep) -> CalendarService:
|
|
return CalendarService(day_override_service)
|
|
|
|
|
|
CalendarServiceDep = Annotated[CalendarService, Depends(get_calendar_service)]
|
|
|
|
|
|
def get_tips_service(
|
|
leave_repo: ParentalLeaveRepositoryDep,
|
|
planned_repo: PlannedLeaveRepositoryDep,
|
|
child_repo: ChildRepositoryDep,
|
|
pd_repo: PersonalDataRepositoryDep,
|
|
vacation_repo: VacationDaysRepositoryDep,
|
|
leave_wish_repo: LeaveWishRepositoryDep,
|
|
ws_repo: WorkspaceRepositoryDep,
|
|
dismissal_repo: TipDismissalRepositoryDep,
|
|
) -> TipsService:
|
|
return TipsService(
|
|
leave_repo,
|
|
planned_repo,
|
|
child_repo,
|
|
pd_repo,
|
|
vacation_repo,
|
|
leave_wish_repo,
|
|
ws_repo,
|
|
dismissal_repo,
|
|
)
|
|
|
|
|
|
TipsServiceDep = Annotated[TipsService, Depends(get_tips_service)]
|
|
|
|
|
|
def get_child_service(repo: ChildRepositoryDep) -> ChildService:
|
|
return ChildService(repo)
|
|
|
|
|
|
ChildServiceDep = Annotated[ChildService, Depends(get_child_service)]
|
|
|
|
|
|
def get_personal_data_service(repo: PersonalDataRepositoryDep) -> PersonalDataService:
|
|
return PersonalDataService(repo)
|
|
|
|
|
|
PersonalDataServiceDep = Annotated[PersonalDataService, Depends(get_personal_data_service)]
|
|
|
|
|
|
def get_workspace_service(
|
|
ws_repo: WorkspaceRepositoryDep,
|
|
inv_repo: WorkspaceInvitationRepositoryDep,
|
|
pd_repo: PersonalDataRepositoryDep,
|
|
user_repo: UserRepositoryDep,
|
|
leave_wish_repo: LeaveWishRepositoryDep,
|
|
planned_leave_repo: PlannedLeaveRepositoryDep,
|
|
parental_leave_repo: ParentalLeaveRepositoryDep,
|
|
vacation_days_repo: VacationDaysRepositoryDep,
|
|
) -> WorkspaceService:
|
|
return WorkspaceService(
|
|
ws_repo,
|
|
inv_repo,
|
|
pd_repo,
|
|
user_repo,
|
|
leave_wish_repo,
|
|
planned_leave_repo,
|
|
parental_leave_repo,
|
|
vacation_days_repo,
|
|
)
|
|
|
|
|
|
WorkspaceServiceDep = Annotated[WorkspaceService, Depends(get_workspace_service)]
|
|
|
|
|
|
def get_analytics_service(
|
|
leave_repo: ParentalLeaveRepositoryDep,
|
|
planned_repo: PlannedLeaveRepositoryDep,
|
|
child_repo: ChildRepositoryDep,
|
|
pd_repo: PersonalDataRepositoryDep,
|
|
workspace_repo: WorkspaceRepositoryDep,
|
|
) -> AnalyticsService:
|
|
return AnalyticsService(leave_repo, planned_repo, child_repo, pd_repo, workspace_repo)
|
|
|
|
|
|
AnalyticsServiceDep = Annotated[AnalyticsService, Depends(get_analytics_service)]
|
|
|
|
|
|
def get_allocation_input_loader(
|
|
workspace_repo: WorkspaceRepositoryDep,
|
|
child_repo: ChildRepositoryDep,
|
|
pd_repo: PersonalDataRepositoryDep,
|
|
parental_leave_repo: ParentalLeaveRepositoryDep,
|
|
vacation_repo: VacationDaysRepositoryDep,
|
|
planned_leave_repo: PlannedLeaveRepositoryDep,
|
|
leave_wish_repo: LeaveWishRepositoryDep,
|
|
day_override_repo: DayOverrideRepositoryDep,
|
|
) -> AllocationInputLoader:
|
|
return AllocationInputLoader(
|
|
workspace_repo=workspace_repo,
|
|
child_repo=child_repo,
|
|
personal_data_repo=pd_repo,
|
|
parental_leave_repo=parental_leave_repo,
|
|
vacation_repo=vacation_repo,
|
|
planned_leave_repo=planned_leave_repo,
|
|
leave_wish_repo=leave_wish_repo,
|
|
day_override_repo=day_override_repo,
|
|
)
|
|
|
|
|
|
AllocationInputLoaderDep = Annotated[AllocationInputLoader, Depends(get_allocation_input_loader)]
|
|
|
|
|
|
def get_planning_engine_service(loader: AllocationInputLoaderDep) -> PlanningEngineService:
|
|
return PlanningEngineService(loader)
|
|
|
|
|
|
PlanningEngineServiceDep = Annotated[PlanningEngineService, Depends(get_planning_engine_service)]
|
|
|
|
|
|
def get_planning_service(
|
|
repo: PlannedLeaveRepositoryDep,
|
|
analytics_service: AnalyticsServiceDep,
|
|
parental_leave_repo: ParentalLeaveRepositoryDep,
|
|
vacation_repo: VacationDaysRepositoryDep,
|
|
planning_engine_service: PlanningEngineServiceDep,
|
|
) -> PlanningService:
|
|
return PlanningService(
|
|
repo, analytics_service, parental_leave_repo, vacation_repo, planning_engine_service
|
|
)
|
|
|
|
|
|
PlanningServiceDep = Annotated[PlanningService, Depends(get_planning_service)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Current-user dependency (JWT bearer)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def get_current_user(
|
|
service: AuthServiceDep,
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
|
|
) -> User:
|
|
"""Extract and verify the JWT bearer token; return the authenticated User."""
|
|
if credentials is None:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Not authenticated")
|
|
try:
|
|
return await service.get_user_from_token(credentials.credentials)
|
|
except UnauthorizedError as exc:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(exc)) from exc
|
|
|
|
|
|
CurrentUserDep = Annotated[User, Depends(get_current_user)]
|