Coverage for packages/server/src/langgate/server/main.py: 71%

1from collections.abc import AsyncIterator 

2from contextlib import asynccontextmanager 

3 

4from fastapi import FastAPI, Request 

5from starlette.middleware.cors import CORSMiddleware 

6from starlette.responses import Response 

7 

8from langgate.core.logging import configure_logger 

9from langgate.server.api.api_v1.api import registry_router 

10from langgate.server.core.config import settings 

11from langgate.server.core.logging import ( 

12 debug_request, 

13 get_logger, 

14 is_debug, 

15 set_structlog_request_context, 

16) 

17 

18configure_logger(json_logs=settings.JSON_LOGS) 

19 

20logger = get_logger(__name__) 

21IS_DEBUG = is_debug() 

22 

23 

24@asynccontextmanager 

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

26 """Handle application startup and shutdown events.""" 

27 logger.info("application_startup") 

28 yield 

29 logger.info("application_shutdown") 

30 

31 

32app = FastAPI( 

33 title=settings.PROJECT_NAME, 

34 openapi_url=f"{settings.API_V1_STR}/openapi.json", 

35 docs_url=f"{settings.API_V1_STR}/docs", 

36 redoc_url=f"{settings.API_V1_STR}/redoc", 

37 lifespan=lifespan, 

38) 

39 

40if settings.CORS_ORIGINS: 

41 app.add_middleware( 

42 CORSMiddleware, 

43 allow_origins=[str(origin) for origin in settings.CORS_ORIGINS], 

44 allow_credentials=True, 

45 allow_methods=["*"], 

46 allow_headers=["*"], 

47 ) 

48 

49 

50@app.middleware("http") 

51async def logging_middleware(request: Request, call_next) -> Response: 

52 set_structlog_request_context(request) 

53 if IS_DEBUG: 

54 await debug_request(logger, request) 

55 response: Response = await call_next(request) 

56 return response 

57 

58 

59app.include_router(registry_router, prefix=settings.API_V1_STR)