Create-knowledge-find-decode/app/dependencies.py

"""FastAPI dependencies for the formal cloud workbench."""
from __future__ import annotations

import os
from typing import Iterator

from fastapi import Depends

from acquisition.repositories.postgres import PostgresAcquisitionRepository
from core.config import CreationDbConfig
from core.db_session import pooled_transaction
from decode_content.repositories.postgres import PostgresDecodeRepository
from pipeline.postgres import PostgresPipelineRepository


def _env_file() -> str:
    return os.getenv("CK_ENV_FILE", ".env")


def get_creation_db_config() -> CreationDbConfig:
    return CreationDbConfig.from_env(_env_file())


def get_db_connection() -> Iterator[object]:
    with pooled_transaction(get_creation_db_config()) as conn:
        yield conn


def get_acquisition_repository(conn: object = Depends(get_db_connection)) -> PostgresAcquisitionRepository:
    return PostgresAcquisitionRepository(conn)


def get_decode_repository(conn: object = Depends(get_db_connection)) -> PostgresDecodeRepository:
    return PostgresDecodeRepository(conn)


def get_pipeline_repository(conn: object = Depends(get_db_connection)) -> PostgresPipelineRepository:
    return PostgresPipelineRepository(conn)