app/services/projects_service.py

import os
import time
from threading import RLock
from typing import List, Optional

from app.repositories.projects_repository import ProjectsRepository
from schemas.projects import Project


class ProjectsService:
    """企画データ取得のサービス層。短時間キャッシュを挟んで DB 負荷を抑える。"""

    def __init__(
        self,
        repository: ProjectsRepository,
        cache_ttl_seconds: Optional[int] = None,
    ) -> None:
        self._repository = repository
        if cache_ttl_seconds is None:
            cache_ttl_seconds = int(os.getenv("PROJECTS_CACHE_TTL", "300"))
        self._cache_ttl = cache_ttl_seconds
        self._cache: Optional[List[Project]] = None
        self._cache_at: Optional[float] = None
        self._lock = RLock()

    def list_projects(self) -> List[Project]:
        """企画一覧を取得する。

        企画データは全てのカラムを含む。
        キャッシュが有効ならそれを返し、無効ならデータベースから取得してキャッシュを更新する。
        """
        now = time.monotonic()
        with self._lock:
            # キャッシュが有効ならそれを返す
            if (
                self._cache is not None
                and self._cache_at is not None
                and now - self._cache_at < self._cache_ttl
            ):
                return [item.model_copy() for item in self._cache]

            # キャッシュが無効ならデータベースから取得してキャッシュを更新する
            projects = [item.model_copy() for item in self._repository.list_projects()]
            self._cache = list(projects)
            self._cache_at = now
            return [item.model_copy() for item in projects]

    def get_project(self, project_id: str) -> Optional[Project]:
        """企画IDから企画データを取得する

        企画データは全てのカラムを含む。
        キャッシュが有効ならそれを返し、無効ならデータベースから取得する。
        """
        if not project_id:
            return None
        now = time.monotonic()
        with self._lock:
            # キャッシュが有効ならそれを返す
            if (
                self._cache is not None
                and self._cache_at is not None
                and now - self._cache_at < self._cache_ttl
            ):
                for item in self._cache:
                    if item.projectId == project_id:
                        return item.model_copy()
                return None

            # キャッシュが無効ならlist_projects経由でデータベースから取得
            projects = self.list_projects()
            for item in projects:
                if item.projectId == project_id:
                    return item.model_copy()
            return None

    def invalidate_cache(self) -> None:
        """キャッシュを無効化する。次回の取得時に必ずデータベースから再取得される。"""
        with self._lock:
            self._cache = None
            self._cache_at = None