Coverage for src/data_api/runtime.py: 100%

1"""Application-owned services and request-local REST/MCP runtime binding.""" 

2 

3from collections.abc import Iterator 

4from contextlib import AsyncExitStack, asynccontextmanager, contextmanager 

5from contextvars import ContextVar 

6from dataclasses import dataclass 

7 

8import httpx 

9from starlette.types import ASGIApp, Receive, Scope, Send 

10 

11from data_api.core.settings import settings 

12from data_api.data.manager import nthudata 

13from data_api.data.nthudata import NTHUDataManager 

14from data_api.domain.announcements.services import AnnouncementsService, announcements_service 

15from data_api.domain.buses.services import BusesService, buses_service 

16from data_api.domain.calendars.services import CalendarsService, calendars_service 

17from data_api.domain.courses.services import CoursesService, courses_service 

18from data_api.domain.departments.services import DepartmentsService, departments_service 

19from data_api.domain.dining.services import DiningService, dining_service 

20from data_api.domain.energy.services import EnergyService, energy_service 

21from data_api.domain.libraries.services import LibrariesService, ctx, libraries_service 

22from data_api.domain.locations.services import LocationsService, locations_service 

23from data_api.domain.newsletters.services import NewslettersService, newsletters_service 

24 

25 

26@dataclass 

27class Runtime: 

28 datasets: NTHUDataManager 

29 buses: BusesService 

30 courses: CoursesService 

31 announcements: AnnouncementsService 

32 calendars: CalendarsService 

33 departments: DepartmentsService 

34 dining: DiningService 

35 energy: EnergyService 

36 libraries: LibrariesService 

37 locations: LocationsService 

38 newsletters: NewslettersService 

39 

40 @classmethod 

41 def create(cls, manager: NTHUDataManager | None = None) -> "Runtime": 

42 manager = manager or NTHUDataManager( 

43 file_details_cache_expiry=settings.file_details_cache_expiry, 

44 http_timeout=settings.data_http_timeout, 

45 ) 

46 return cls( 

47 manager, 

48 BusesService(manager), 

49 CoursesService(manager), 

50 AnnouncementsService(manager), 

51 CalendarsService(manager), 

52 DepartmentsService(manager), 

53 DiningService(manager), 

54 EnergyService(), 

55 LibrariesService(manager), 

56 LocationsService(manager), 

57 NewslettersService(manager), 

58 ) 

59 

60 @asynccontextmanager 

61 async def lifespan(self): 

62 if self.datasets.fetcher.client is not None: 

63 raise RuntimeError("Runtime lifespan already running") 

64 async with AsyncExitStack() as stack: 

65 energy_client = await stack.enter_async_context(httpx.AsyncClient()) 

66 library_client = await stack.enter_async_context(httpx.AsyncClient(verify=ctx)) 

67 self.energy.client = energy_client 

68 self.libraries.client = library_client 

69 try: 

70 await stack.enter_async_context(self.datasets.lifespan()) 

71 yield 

72 finally: 

73 self.energy.client = None 

74 self.libraries.client = None 

75 self.buses.reset() 

76 

77 

78default_runtime = Runtime( 

79 nthudata, 

80 buses_service, 

81 courses_service, 

82 announcements_service, 

83 calendars_service, 

84 departments_service, 

85 dining_service, 

86 energy_service, 

87 libraries_service, 

88 locations_service, 

89 newsletters_service, 

90) 

91_active_runtime: ContextVar[Runtime | None] = ContextVar("data_api_runtime", default=None) 

92 

93 

94def get_runtime() -> Runtime: 

95 runtime = _active_runtime.get() 

96 return default_runtime if runtime is None else runtime 

97 

98 

99@contextmanager 

100def bind_runtime(runtime: Runtime) -> Iterator[None]: 

101 token = _active_runtime.set(runtime) 

102 try: 

103 yield 

104 finally: 

105 _active_runtime.reset(token) 

106 

107 

108class RuntimeMiddleware: 

109 def __init__(self, app: ASGIApp, runtime: Runtime): 

110 self.app = app 

111 self.runtime = runtime 

112 

113 async def __call__(self, scope: Scope, receive: Receive, send: Send): 

114 with bind_runtime(self.runtime): 

115 await self.app(scope, receive, send)