fix: P2-4 add complete type annotations to dependencies
Deploy / Deploy Staging (push) Failing after 1s
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 1m1s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 1m0s
Tests / test (pull_request) Failing after 1m0s
Tests / lint (pull_request) Failing after 31s

This commit is contained in:
2026-06-26 17:58:25 +08:00
parent cf02415e9e
commit 712b01d805
+35 -3
View File
@@ -1,4 +1,13 @@
"""Dependency injection providers for FastAPI endpoints.
All repository and service factories are defined here as FastAPI dependencies,
ensuring proper lifecycle management and testability.
"""
from __future__ import annotations
import redis
from typing import Generator
from app.config import settings
from fastapi import Depends
from sqlalchemy.orm import Session
@@ -32,15 +41,24 @@ from packages.adapters.sqlalchemy_impl.user_repository import SQLAlchemyUserRepo
from packages.adapters.sqlalchemy_impl.workspace_invitation_repository import SQLAlchemyWorkspaceInvitationRepository
from packages.adapters.sqlalchemy_impl.workspace_member_repository import SQLAlchemyWorkspaceMemberRepository
from packages.adapters.sqlalchemy_impl.workspace_repository import SQLAlchemyWorkspaceRepository
from packages.ports.asset_repository import AssetRepository
from packages.ports.asset_library_repository import AssetLibraryRepository
from packages.ports.user_repository import UserRepository
from packages.ports.workspace_invitation_repository import WorkspaceInvitationRepository
from packages.ports.workspace_member_repository import WorkspaceMemberRepository
from packages.ports.workspace_repository import WorkspaceRepository
from packages.ports.classification_job_repository import ClassificationJobRepository
from packages.ports.generation_task_repository import GenerationTaskRepository
from packages.ports.generated_video_repository import GeneratedVideoRepository
from packages.ports.ingest_job_repository import IngestJobRepository
from packages.ports.project_repository import ProjectRepository
from packages.ports.project_title_repository import ProjectTitleRepository
_engine, _SessionLocal = build_session_factory(settings.DATABASE_URL)
def get_db_session():
def get_db_session() -> Generator[Session, None, None]:
"""Provide a database session with automatic cleanup."""
session: Session = _SessionLocal()
try:
yield session
@@ -51,82 +69,96 @@ def get_db_session():
def get_asset_repository(
session: Session = Depends(get_db_session),
) -> SQLAlchemyAssetRepository:
"""Provide the SQLAlchemy asset repository implementation."""
return SQLAlchemyAssetRepository(session)
def get_asset_library_repository(
session: Session = Depends(get_db_session),
) -> SQLAlchemyAssetLibraryRepository:
"""Provide the SQLAlchemy asset library repository implementation."""
return SQLAlchemyAssetLibraryRepository(session)
def get_ingest_job_repository(
session: Session = Depends(get_db_session),
) -> SQLAlchemyIngestJobRepository:
"""Provide the SQLAlchemy ingest job repository implementation."""
return SQLAlchemyIngestJobRepository(session)
def get_classification_job_repository(
session: Session = Depends(get_db_session),
) -> SQLAlchemyClassificationJobRepository:
"""Provide the SQLAlchemy classification job repository implementation."""
return SQLAlchemyClassificationJobRepository(session)
def get_generation_task_repository(
session: Session = Depends(get_db_session),
) -> SQLAlchemyGenerationTaskRepository:
"""Provide the SQLAlchemy generation task repository implementation."""
return SQLAlchemyGenerationTaskRepository(session)
def get_generated_video_repository(
session: Session = Depends(get_db_session),
) -> SQLAlchemyGeneratedVideoRepository:
"""Provide the SQLAlchemy generated video repository implementation."""
return SQLAlchemyGeneratedVideoRepository(session)
def get_project_repository(
session: Session = Depends(get_db_session),
) -> SQLAlchemyProjectRepository:
"""Provide the SQLAlchemy project repository implementation."""
return SQLAlchemyProjectRepository(session)
def get_project_title_repository(
session: Session = Depends(get_db_session),
) -> SQLAlchemyProjectTitleRepository:
"""Provide the SQLAlchemy project title repository implementation."""
return SQLAlchemyProjectTitleRepository(session)
def get_user_repository(
session: Session = Depends(get_db_session),
) -> UserRepository:
"""Provide the SQLAlchemy user repository implementation."""
return SQLAlchemyUserRepository(session)
def get_workspace_repository(
session: Session = Depends(get_db_session),
) -> WorkspaceRepository:
"""Provide the SQLAlchemy workspace repository implementation."""
return SQLAlchemyWorkspaceRepository(session)
def get_workspace_member_repository(
session: Session = Depends(get_db_session),
) -> WorkspaceMemberRepository:
"""Provide the SQLAlchemy workspace member repository implementation."""
return SQLAlchemyWorkspaceMemberRepository(session)
def get_workspace_invitation_repository(
session: Session = Depends(get_db_session),
) -> WorkspaceInvitationRepository:
"""Provide the SQLAlchemy workspace invitation repository implementation."""
return SQLAlchemyWorkspaceInvitationRepository(session)
def get_auth_session_store():
def get_auth_session_store() -> SessionStore | NoopSessionStore:
"""Provide the session store based on configuration."""
if not settings.ENABLE_REDIS_SESSIONS:
return NoopSessionStore()
return SessionStore(redis_client=redis.from_url(settings.REDIS_URL, decode_responses=True))
def get_auth_email_service():
def get_auth_email_service() -> NoopEmailService | EmailService:
"""Provide the email service based on configuration."""
if not settings.ENABLE_EMAIL_DELIVERY:
return NoopEmailService()
return get_email_service(