-
Notifications
You must be signed in to change notification settings - Fork 3
/
home.py
40 lines (31 loc) · 1.42 KB
/
home.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from dependencies import circuit_breaker
from fastapi import APIRouter, HTTPException
from fastapi.responses import JSONResponse
from app.config.base import settings
from app.middlewares.request_id_injection import request_id_contextvar
from app.daos.home import external_service_call
from pybreaker import CircuitBreakerError
home_router = APIRouter()
@home_router.get("/", tags=["Home"])
async def read_main():
print("Request ID:", request_id_contextvar.get())
return {"response": "service up and running..!"}
@home_router.get("/external-service", tags=["Home"])
async def external_service_endpoint():
try:
with circuit_breaker.calling():
result = await external_service_call()
return {"message": result}
except CircuitBreakerError:
raise HTTPException(status_code=503, detail="Service temporarily unavailable")
except Exception as e:
# Handle other exceptions from the external service call
raise HTTPException(status_code=500, detail=str(e))
@home_router.get("/sentry-test", tags=["Home"])
def sentry_endpoint():
if not settings.SENTRY_DSN:
raise HTTPException(status_code=503, detail="Sentry DSN not found")
raise Exception("Testing Sentry")
@home_router.get("/{path:path}", include_in_schema=False)
async def catch_all(path: str):
return JSONResponse(status_code=404, content={"success": False, "message": f"Route not found for path: {path}"})