Spaces:
Sleeping
Sleeping
File size: 3,146 Bytes
67c9f93 351a26a 67c9f93 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | 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
|