-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
202 lines (161 loc) · 6.04 KB
/
main.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
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
import logging
from typing import Optional
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Depends, status
from fastapi.responses import RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from routers import account, dashboard, organization, role, user, static_pages
from utils.dependencies import (
get_optional_user
)
from exceptions.http_exceptions import (
AuthenticationError,
PasswordValidationError
)
from exceptions.exceptions import (
NeedsNewTokens
)
from utils.db import set_up_db
from utils.models import User
logger = logging.getLogger("uvicorn.error")
logger.setLevel(logging.DEBUG)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Optional startup logic
set_up_db()
yield
# Optional shutdown logic
# Initialize the FastAPI app
app: FastAPI = FastAPI(lifespan=lifespan)
# Mount static files (e.g., CSS, JS) and initialize Jinja2 templates
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
# --- Include Routers ---
app.include_router(account.router)
app.include_router(dashboard.router)
app.include_router(organization.router)
app.include_router(role.router)
app.include_router(static_pages.router)
app.include_router(user.router)
# --- Exception Handling Middlewares ---
# Handle AuthenticationError by redirecting to login page
@app.exception_handler(AuthenticationError)
async def authentication_error_handler(request: Request, exc: AuthenticationError):
return RedirectResponse(
url=app.url_path_for("read_login"),
status_code=status.HTTP_303_SEE_OTHER
)
# Handle NeedsNewTokens by setting new tokens and redirecting to same page
@app.exception_handler(NeedsNewTokens)
async def needs_new_tokens_handler(request: Request, exc: NeedsNewTokens):
response = RedirectResponse(
url=request.url.path, status_code=status.HTTP_307_TEMPORARY_REDIRECT)
response.set_cookie(
key="access_token",
value=exc.access_token,
httponly=True,
secure=True,
samesite="strict"
)
response.set_cookie(
key="refresh_token",
value=exc.refresh_token,
httponly=True,
secure=True,
samesite="strict"
)
return response
# Handle PasswordValidationError by rendering the validation_error page
@app.exception_handler(PasswordValidationError)
async def password_validation_exception_handler(request: Request, exc: PasswordValidationError):
return templates.TemplateResponse(
request,
"errors/validation_error.html",
{
"status_code": 422,
"errors": {"error": exc.detail}
},
status_code=422,
)
# Handle RequestValidationError by rendering the validation_error page
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = {}
# Map error types to user-friendly message templates
error_templates = {
"pattern_mismatch": "this field cannot be empty or contain only whitespace",
"string_too_short": "this field is required",
"missing": "this field is required",
"string_pattern_mismatch": "this field cannot be empty or contain only whitespace",
"enum": "invalid value"
}
for error in exc.errors():
# Handle different error locations carefully
location = error["loc"]
# Skip type errors for the whole body
if len(location) == 1 and location[0] == "body":
continue
# For form fields, the location might be just (field_name,)
# For JSON body, it might be (body, field_name)
# For array items, it might be (field_name, array_index)
field_name = location[-2] if isinstance(location[-1], int) else location[-1]
# Format the field name to be more user-friendly
display_name = field_name.replace("_", " ").title()
# Use mapped message if available, otherwise use FastAPI's message
error_type = error.get("type", "")
message_template = error_templates.get(error_type, error["msg"])
# For array items, append the index to the message
if isinstance(location[-1], int):
message_template = f"Item {location[-1] + 1}: {message_template}"
errors[display_name] = message_template
return templates.TemplateResponse(
request,
"errors/validation_error.html",
{
"status_code": 422,
"errors": errors
},
status_code=422,
)
# Handle StarletteHTTPException (including 404, 405, etc.) by rendering the error page
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return templates.TemplateResponse(
request,
"errors/error.html",
{"status_code": exc.status_code, "detail": exc.detail},
status_code=exc.status_code,
)
# Add handler for uncaught exceptions (500 Internal Server Error)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
# Log the error for debugging
logger.error(f"Unhandled exception: {exc}", exc_info=True)
return templates.TemplateResponse(
request,
"errors/error.html",
{
"status_code": 500,
"detail": "Internal Server Error"
},
status_code=500,
)
# --- Home Page ---
@app.get("/")
async def read_home(
request: Request,
user: Optional[User] = Depends(get_optional_user)
):
if user:
return RedirectResponse(url=app.url_path_for("read_dashboard"), status_code=302)
return templates.TemplateResponse(
request,
"index.html",
{"user": user}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)