-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest.py
More file actions
138 lines (123 loc) · 3.85 KB
/
test.py
File metadata and controls
138 lines (123 loc) · 3.85 KB
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
import logging
import traceback
from protogrid import make_response
# ---------------- Logging ----------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s"
)
app = FastAPI()
# ---------------- CORS ----------------
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/bhai")
def root():
try:
response = make_response(
payload=data,
message="User retrieved successfully"
)
return response
except Exception as e:
return make_response(
status=500,
message="Internal Server Error",
payload=str(e),
error_details="API bhai is not working",
include_meta=True
)
# ======================================================
# HEALTH CHECK API (to check if server is running)
# ======================================================
@app.get("/health")
def health_check():
return make_response(
status=200,
message="Server is running",
payload={"status": "OK"},
include_meta=True
)
# ======================================================
# TEST SUCCESS API
# ======================================================
@app.get("/success")
def success_api():
return make_response(
status=200,
message="Success API working",
page=1,
limit=10,
total_items=10,
include_meta=True,
error=""
)
# ======================================================
# TEST ERROR API (manual error trigger)
# ======================================================
@app.get("/error")
def error_api():
return make_response(
status=400,
message="Manual Bad Request",
page=1,
limit=10,
total_items=10,
include_meta=True,
error_details="Manual Bad Request"
)
# ======================================================
# 404 NOT FOUND HANDLER
# ======================================================
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
logging.error(f"HTTP Error: {exc.detail}")
return JSONResponse(
status_code=exc.status_code,
content=make_response(
status=exc.status_code,
message="HTTP Error",
payload=exc.detail,
include_meta=False
)
)
# ======================================================
# VALIDATION ERROR HANDLER (422)
# ======================================================
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
logging.error(f"Validation Error: {exc.errors()}")
return JSONResponse(
status_code=422,
content=make_response(
status=422,
message="Validation Error",
payload=exc.errors(),
include_meta=False
)
)
# ======================================================
# GLOBAL ERROR HANDLER (500)
# ======================================================
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logging.error(f"Unhandled Error: {str(exc)}")
logging.error(traceback.format_exc())
return JSONResponse(
status_code=500,
content=make_response(
status=500,
message="Internal Server Error",
payload=str(exc),
include_meta=False
)
)