xiaoxia-saas/apps/api/app/dependencies.py

from typing import Generator

import redis
from app.config import settings
from fastapi import Depends
from sqlalchemy.orm import Session

from packages.adapters.redis import NoopSessionStore, SessionStore
from packages.adapters.smtp import EmailConfig, NoopEmailService, get_email_service
from packages.adapters.sqlalchemy_impl.asset_library_repository import (
    SQLAlchemyAssetLibraryRepository,
)
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.adapters.sqlalchemy_impl.classification_job_repository import (
    SQLAlchemyClassificationJobRepository,
)
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
    SQLAlchemyGeneratedVideoRepository,
)
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
    SQLAlchemyGenerationTaskRepository,
)
from packages.adapters.sqlalchemy_impl.ingest_job_repository import (
    SQLAlchemyIngestJobRepository,
)
from packages.adapters.sqlalchemy_impl.project_repository import (
    SQLAlchemyProjectRepository,
)
from packages.adapters.sqlalchemy_impl.project_title_repository import (
    SQLAlchemyProjectTitleRepository,
)
from packages.adapters.sqlalchemy_impl.session import build_session_factory
from packages.adapters.sqlalchemy_impl.user_repository import SQLAlchemyUserRepository
from packages.ports.user_repository import UserRepository

_engine, _SessionLocal = build_session_factory(settings.DATABASE_URL)


def get_db_session() -> Generator[Session, None, None]:
    """获取数据库会话"""
    session: Session = _SessionLocal()
    try:
        yield session
    finally:
        session.close()


def get_asset_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyAssetRepository:
    return SQLAlchemyAssetRepository(session)


def get_asset_library_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyAssetLibraryRepository:
    return SQLAlchemyAssetLibraryRepository(session)


def get_ingest_job_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyIngestJobRepository:
    return SQLAlchemyIngestJobRepository(session)


def get_classification_job_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyClassificationJobRepository:
    return SQLAlchemyClassificationJobRepository(session)


def get_generation_task_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyGenerationTaskRepository:
    return SQLAlchemyGenerationTaskRepository(session)


def get_generated_video_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyGeneratedVideoRepository:
    return SQLAlchemyGeneratedVideoRepository(session)


def get_project_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyProjectRepository:
    return SQLAlchemyProjectRepository(session)


def get_project_title_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyProjectTitleRepository:
    return SQLAlchemyProjectTitleRepository(session)


def get_user_repository(
    session: Session = Depends(get_db_session),
) -> UserRepository:
    return SQLAlchemyUserRepository(session)


def get_auth_session_store() -> NoopSessionStore | SessionStore:
    if not settings.ENABLE_REDIS_SESSION:
        return NoopSessionStore()
    return SessionStore(redis_client=redis.from_url(settings.REDIS_URL, decode_responses=True))


def get_auth_email_service() -> NoopEmailService:
    if not settings.ENABLE_EMAIL_DELIVERY:
        return NoopEmailService()
    return get_email_service(
        EmailConfig(
            smtp_host=settings.SMTP_HOST,
            smtp_port=settings.SMTP_PORT,
            smtp_user=settings.SMTP_USER,
            smtp_password=settings.SMTP_PASSWORD,
            from_email=settings.SMTP_FROM_EMAIL,
            from_name=settings.SMTP_FROM_NAME,
            use_tls=settings.SMTP_USE_TLS,
        ),
        enabled=True,
    )