Coverage for src/cstlcore/main.py: 87%

1from contextlib import asynccontextmanager 

2 

3from fastapi import FastAPI 

4from fastapi.middleware.cors import CORSMiddleware 

5from fastapi.responses import JSONResponse 

6from sqlalchemy.exc import IntegrityError 

7 

8from cstlcore.assets import router as assets 

9from cstlcore.auth import router as auth 

10from cstlcore.collections import router as collections 

11from cstlcore.constellations import router as constellations 

12from cstlcore.feedbacks import router as feedbacks 

13from cstlcore.glossary import router as glossary 

14from cstlcore.logging import setup_logging 

15from cstlcore.maps import router as maps 

16from cstlcore.me import router as me 

17from cstlcore.memberships import router as memberships 

18from cstlcore.newsletter import router as newsletter 

19from cstlcore.settings import settings 

20from cstlcore.users import router as users 

21from cstlcore.ydocs import router as ydocs 

22from cstlcore.admin import router as admin 

23 

24setup_logging() 

25 

26 

27@asynccontextmanager 

28async def lifespan(app: FastAPI): 

29 # On startup ↑ 

30 yield 

31 # On shutdown ↓ 

32 

33 

34app = FastAPI(lifespan=lifespan) 

35 

36 

37@app.exception_handler(IntegrityError) 

38async def sqlalchemy_integrity_error_handler(request, exc: IntegrityError): 

39 """Convert DB integrity errors (unique constraint, etc.) to HTTP 409.""" 

40 # We return a simple message; more detailed parsing could extract which 

41 # constraint failed (e.g., unique vs. check) but that's optional. 

42 return JSONResponse(status_code=409, content={"detail": "Database integrity error"}) 

43 

44 

45app.add_middleware( 

46 CORSMiddleware, 

47 allow_origins=settings.allowed_origins, 

48 allow_credentials=True, 

49 allow_methods=["*"], 

50 allow_headers=["*"], 

51) 

52 

53app.include_router(constellations.router, tags=["Constellations"]) 

54app.include_router(collections.router, tags=["Collections"]) 

55app.include_router(users.router, tags=["Users"]) 

56app.include_router(memberships.router, tags=["Memberships"]) 

57app.include_router(auth.router, tags=["Authentication"]) 

58app.include_router(ydocs.router, tags=["YDocs"]) 

59app.include_router(me.router, tags=["Me"]) 

60app.include_router(assets.router, tags=["Assets"]) 

61app.include_router(admin.router, tags=["Admin"]) 

62app.include_router(maps.router, tags=["Maps"]) 

63app.include_router(newsletter.router, tags=["Newsletter"]) 

64app.include_router(feedbacks.router, tags=["Feedbacks"]) 

65app.include_router(glossary.router, tags=["Glossary"]) 

66 

67if settings.target == "dev": 

68 from cstlcore.debug import router as debug 

69 

70 app.include_router(debug.router, tags=["Debug"]) 

71 

72 

73@app.get("/health") 

74async def health_check(): 

75 return {"status": "ok"} 

76 

77 

78def main(): 

79 import uvicorn 

80 

81 uvicorn.run( 

82 app, 

83 host="0.0.0.0", 

84 port=8001, 

85 ) 

86 

87 

88if __name__ == "__main__": 

89 main()