Coverage for src/data_api/api/api.py: 96%

1""" 

2FastAPI application setup. 

3 

4Creates the FastAPI app instance, configures middleware, 

5and registers all routers. 

6""" 

7 

8import logging 

9import time 

10from contextlib import asynccontextmanager 

11 

12from fastapi import FastAPI, Request 

13from fastapi.middleware.cors import CORSMiddleware 

14from fastapi.responses import JSONResponse 

15from starlette.types import ASGIApp, Message, Receive, Scope, Send 

16 

17from data_api import __version__ 

18from data_api.api.schemas.meta import MetaResponse 

19from data_api.core import config 

20from data_api.core.exceptions import ( 

21 INTERNAL_ERROR_DETAIL, 

22 DataNotAvailableException, 

23 UpstreamException, 

24) 

25from data_api.core.settings import settings 

26from data_api.mcp import mcp 

27from data_api.mcp.server import create_mcp 

28from data_api.runtime import Runtime, RuntimeMiddleware, default_runtime 

29 

30logger = logging.getLogger(__name__) 

31 

32 

33@asynccontextmanager 

34async def lifespan(app: FastAPI): 

35 """Application lifespan manager for startup and shutdown tasks.""" 

36 runtime: Runtime = app.state.runtime 

37 async with runtime.lifespan(): 

38 app.state.datasets = runtime.datasets 

39 results = await runtime.datasets.prefetch(config.PREFETCH_ENDPOINTS) 

40 logging.getLogger(__name__).info( 

41 "Data startup: %s/%s datasets usable", sum(results.values()), len(results) 

42 ) 

43 for update in (runtime.buses.update_data, runtime.courses.get_snapshot): 

44 try: 

45 await update() 

46 except DataNotAvailableException: 

47 # The manager records the failure. Other datasets must remain available. 

48 continue 

49 yield 

50 

51 

52# MCP's session manager must run alongside the data initialization lifespan. 

53mcp_app = mcp.http_app(path="/mcp", transport="streamable-http", stateless_http=True) 

54 

55 

56@asynccontextmanager 

57async def combined_lifespan(app: FastAPI): 

58 """Run both MCP and data lifespans on the exported application.""" 

59 transport = mcp_app if app.state.runtime is default_runtime else app.state.mcp_app 

60 async with transport.lifespan(app), lifespan(app): 

61 yield 

62 

63 

64class _SafeUnexpectedErrorsMiddleware: 

65 def __init__(self, app: ASGIApp): 

66 self.app = app 

67 

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

69 if scope["type"] != "http": 

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

71 return 

72 

73 response_started = False 

74 

75 async def track_response(message: Message): 

76 nonlocal response_started 

77 if message["type"] == "http.response.start": 

78 response_started = True 

79 await send(message) 

80 

81 try: 

82 await self.app(scope, receive, track_response) 

83 except Exception as exc: 

84 if response_started: 

85 raise 

86 logger.error("Unexpected request failure", exc_info=exc) 

87 await JSONResponse(status_code=500, content={"detail": INTERNAL_ERROR_DETAIL})( 

88 scope, receive, send 

89 ) 

90 

91 

92def create_app(runtime: Runtime | None = None) -> FastAPI: 

93 """ 

94 Create and configure the FastAPI application. 

95 

96 Returns: 

97 FastAPI: Configured application instance. 

98 """ 

99 runtime = Runtime.create() if runtime is None else runtime 

100 server = mcp if runtime is default_runtime else create_mcp(runtime) 

101 transport = ( 

102 mcp_app 

103 if runtime is default_runtime 

104 else server.http_app(path="/mcp", transport="streamable-http", stateless_http=True) 

105 ) 

106 app = FastAPI( 

107 lifespan=combined_lifespan, 

108 routes=list(transport.routes), 

109 title="NTHU Data API", 

110 version=__version__, 

111 description="由國立清華大學校內各單位資料所組成的公共資料 API。", 

112 ) 

113 app.state.runtime = runtime 

114 app.state.datasets = runtime.datasets 

115 app.state.mcp = server 

116 app.state.mcp_app = transport 

117 app.add_middleware(RuntimeMiddleware, runtime=runtime) 

118 

119 @app.exception_handler(DataNotAvailableException) 

120 async def dataset_unavailable(request: Request, exc: DataNotAvailableException): 

121 return JSONResponse(status_code=503, content={"detail": "Service temporarily unavailable"}) 

122 

123 @app.exception_handler(UpstreamException) 

124 async def upstream_failure(request: Request, exc: UpstreamException): 

125 logger.warning("Live upstream request failed", exc_info=exc) 

126 return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail}) 

127 

128 @app.exception_handler(Exception) 

129 async def internal_error(request: Request, exc: Exception): 

130 logger.error("Unexpected request failure", exc_info=exc) 

131 return JSONResponse(status_code=500, content={"detail": INTERNAL_ERROR_DETAIL}) 

132 

133 # CORS and timing wrap this fallback so pre-response errors keep their headers. 

134 app.add_middleware(_SafeUnexpectedErrorsMiddleware) 

135 

136 # CORS configuration 

137 # Using explicit origins would be safer, but for a public API: 

138 origins = settings.cors_origins # From settings 

139 

140 app.add_middleware( 

141 CORSMiddleware, 

142 allow_origins=origins, 

143 allow_credentials=False, # False when using wildcard origins 

144 allow_methods=[ 

145 "GET", 

146 "HEAD", 

147 "POST", 

148 "PUT", 

149 "DELETE", 

150 "OPTIONS", 

151 "PATCH", 

152 ], 

153 allow_headers=["*"], 

154 expose_headers=[ 

155 "X-Process-Time", 

156 "X-Data-Commit-Hash", 

157 "X-Total-Count", 

158 ], 

159 ) 

160 

161 # Process time middleware 

162 @app.middleware("http") 

163 async def add_process_time_header(request: Request, call_next): 

164 start_time = time.perf_counter() 

165 response = await call_next(request) 

166 process_time = time.perf_counter() - start_time 

167 response.headers["X-Process-Time"] = str(process_time) 

168 return response 

169 

170 @app.get("/", include_in_schema=False) 

171 async def metadata() -> MetaResponse: 

172 """Discover the API name, version, documentation, and MCP endpoint.""" 

173 return MetaResponse( 

174 name=app.title, 

175 version=app.version, 

176 docs="/docs", 

177 openapi="/openapi.json", 

178 mcp="/mcp", 

179 ) 

180 

181 # Add favicon route 

182 @app.get("/favicon.ico", include_in_schema=False) 

183 async def favicon(): 

184 """Favicon route.""" 

185 from fastapi.responses import RedirectResponse 

186 

187 return RedirectResponse(url="https://www.nthu.edu.tw/favicon.ico") 

188 

189 # Register routers 

190 from data_api.api.routers import ( 

191 announcements, 

192 buses, 

193 calendars, 

194 courses, 

195 dining, 

196 directory, 

197 energy, 

198 libraries, 

199 locations, 

200 newsletters, 

201 ping, 

202 ) 

203 

204 app.include_router(ping.router) 

205 app.include_router(announcements.router, prefix="/announcements", tags=["Announcements"]) 

206 app.include_router(buses.router, prefix="/buses", tags=["Buses"]) 

207 app.include_router(calendars.router, prefix="/calendars", tags=["Calendars"]) 

208 app.include_router(courses.router, prefix="/courses", tags=["Courses"]) 

209 app.include_router(directory.router, prefix="/directory", tags=["Directory"]) 

210 app.include_router(dining.router, prefix="/dining", tags=["Dining"]) 

211 app.include_router(energy.router, prefix="/energy", tags=["Energy"]) 

212 app.include_router(libraries.router, prefix="/libraries", tags=["Libraries"]) 

213 app.include_router(locations.router, prefix="/locations", tags=["Locations"]) 

214 app.include_router(newsletters.router, prefix="/newsletters", tags=["Newsletters"]) 

215 

216 return app 

217 

218 

219app = create_app(default_runtime) 

220 

221# Preserve module aliases without constructing a second, unconfigured app. 

222fast_api_app = combined_app = app