Coverage for src / mcp_server_langgraph / infrastructure / app_factory.py: 57%

1""" 

2FastAPI Application Factory 

3 

4This module provides factory functions for creating FastAPI applications 

5with proper configuration, middleware, and lifecycle management. 

6 

7Separates infrastructure concerns from business logic. 

8""" 

9 

10from __future__ import annotations 

11 

12import logging 

13from collections.abc import AsyncIterator 

14from contextlib import asynccontextmanager 

15from typing import Any 

16 

17from fastapi import FastAPI 

18from fastapi.middleware.cors import CORSMiddleware 

19 

20from mcp_server_langgraph.core.config import Settings 

21from mcp_server_langgraph.core.container import ApplicationContainer, create_test_container 

22 

23logger = logging.getLogger(__name__) 

24 

25 

26def create_app( 

27 container: ApplicationContainer | None = None, 

28 settings: Settings | None = None, 

29 environment: str | None = None, 

30) -> FastAPI: 

31 """ 

32 Create a FastAPI application with proper configuration. 

33 

34 This factory function creates a FastAPI app with: 

35 - Proper middleware (CORS, rate limiting, auth) 

36 - Lifecycle management (startup/shutdown) 

37 - OpenAPI customization 

38 - Health check endpoints 

39 

40 Args: 

41 container: Optional ApplicationContainer for DI 

42 settings: Optional Settings (if container not provided) 

43 environment: Optional environment override 

44 

45 Returns: 

46 Configured FastAPI application 

47 

48 Example: 

49 # Using container (preferred) 

50 from mcp_server_langgraph.core.container import create_test_container 

51 container = create_test_container() 

52 app = create_app(container=container) 

53 

54 # Using custom settings 

55 from mcp_server_langgraph.core.config import Settings 

56 settings = Settings(service_name="my-service") 

57 app = create_app(settings=settings) 

58 

59 # Using defaults 

60 app = create_app() 

61 """ 

62 # Get or create container 

63 if container is None: 

64 if environment == "test" or (settings and settings.environment == "test"): 

65 container = create_test_container(settings=settings) 

66 else: 

67 from mcp_server_langgraph.core.container import ApplicationContainer, ContainerConfig 

68 

69 env = environment or (settings.environment if settings else "development") 

70 config = ContainerConfig(environment=env) 

71 container = ApplicationContainer(config, settings=settings) 

72 

73 # Get settings from container 

74 app_settings = container.settings 

75 

76 # Create lifespan wrapper to inject container 

77 @asynccontextmanager 

78 async def lifespan(app: FastAPI) -> AsyncIterator[None]: 

79 async with create_lifespan(container=container): 

80 yield 

81 

82 # Create FastAPI app 

83 app = FastAPI( 

84 title=app_settings.service_name, 

85 description="MCP Server with LangGraph", 

86 version="1.0.0", 

87 lifespan=lifespan, 

88 ) 

89 

90 # Add CORS middleware 

91 app.add_middleware( 

92 CORSMiddleware, 

93 allow_origins=["*"], # Configure based on settings in production 

94 allow_credentials=True, 

95 allow_methods=["*"], 

96 allow_headers=["*"], 

97 ) 

98 

99 # Add health check endpoint 

100 @app.get("/health") 

101 async def health_check() -> dict[str, str]: 

102 """Health check endpoint""" 

103 return {"status": "healthy", "service": app_settings.service_name} 

104 

105 # Customize OpenAPI 

106 app.openapi_schema = None # Reset to trigger regeneration 

107 

108 return app 

109 

110 

111@asynccontextmanager 

112async def create_lifespan(container: ApplicationContainer | None = None) -> AsyncIterator[None]: 

113 """ 

114 Create application lifespan context manager. 

115 

116 Handles startup and shutdown tasks: 

117 - Initialize telemetry 

118 - Setup connections 

119 - Cleanup resources 

120 

121 Args: 

122 container: Optional ApplicationContainer 

123 

124 Yields: 

125 None (context manager pattern) 

126 

127 Example: 

128 lifespan = create_lifespan(container) 

129 app = FastAPI(lifespan=lifespan) 

130 """ 

131 # Startup 

132 if container: 

133 _telemetry = container.get_telemetry() # noqa: F841 

134 logger.info(f"Application starting (environment: {container.settings.environment})") 

135 

136 # Validate checkpoint configuration at startup (fail-fast) 

137 from mcp_server_langgraph.core.checkpoint_validator import validate_checkpoint_config 

138 

139 try: 

140 validate_checkpoint_config(container.settings) 

141 except Exception as e: 

142 logger.exception(f"Checkpoint configuration validation failed: {e}") 

143 # Re-raise to prevent application from starting with invalid config 

144 raise 

145 

146 # Initialize GDPR storage at startup (fail-fast) 

147 from mcp_server_langgraph.compliance.gdpr.factory import initialize_gdpr_storage 

148 

149 try: 

150 logger.info(f"Initializing GDPR storage (backend: {container.settings.gdpr_storage_backend})") 

151 await initialize_gdpr_storage( 

152 backend=container.settings.gdpr_storage_backend, # type: ignore[arg-type] 

153 postgres_url=container.settings.gdpr_postgres_url, 

154 ) 

155 logger.info("GDPR storage initialized successfully") 

156 except Exception as e: 

157 logger.exception(f"GDPR storage initialization failed: {e}") 

158 # Re-raise to prevent application from starting without GDPR storage 

159 raise 

160 

161 yield 

162 

163 # Shutdown 

164 if container: 

165 logger.info("Application shutting down") 

166 

167 # Reset GDPR storage on shutdown 

168 from mcp_server_langgraph.compliance.gdpr.factory import reset_gdpr_storage 

169 

170 reset_gdpr_storage() 

171 logger.info("GDPR storage reset") 

172 

173 

174def customize_openapi(app: FastAPI) -> dict[str, Any]: 

175 """ 

176 Customize OpenAPI schema for the application. 

177 

178 Args: 

179 app: FastAPI application 

180 

181 Returns: 

182 Customized OpenAPI schema dict 

183 

184 Example: 

185 app = FastAPI() 

186 schema = customize_openapi(app) 

187 app.openapi_schema = schema 

188 """ 

189 from typing import cast 

190 

191 if app.openapi_schema: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true

192 return cast(dict[str, Any], app.openapi_schema) # type: ignore[redundant-cast] 

193 

194 from fastapi.openapi.utils import get_openapi 

195 

196 openapi_schema = get_openapi( 

197 title=app.title, 

198 version=app.version, 

199 description=app.description, 

200 routes=app.routes, 

201 ) 

202 

203 # Add custom properties 

204 openapi_schema["info"]["x-logo"] = {"url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png"} 

205 

206 app.openapi_schema = openapi_schema 

207 return cast(dict[str, Any], app.openapi_schema) # type: ignore[redundant-cast]