-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
83 lines (67 loc) · 2.63 KB
/
main.py
File metadata and controls
83 lines (67 loc) · 2.63 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
# main.py
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from fastapi.responses import RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from app import routes, database, logger, auth
from sqlalchemy.orm import Session
from app.database import get_session_local
from app.crud import get_user_by_email
from app.auth import verify_token
app = FastAPI(
title="User Manager MS",
description="API for Auth",
version="1.0.0"
)
# Add your custom OpenAPI schema here if needed
security = HTTPBearer()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], #
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.on_event("startup")
def startup():
database.init_db()
logger.log_message("Database initialized.")
app.include_router(routes.router)
@app.get("/", include_in_schema=False)
def index():
return RedirectResponse(url="/login", status_code=303)
# Routing for the dashboard page
@app.get("/store", include_in_schema=False, response_class=HTMLResponse)
def dashboard_page(request: Request, credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_session_local)):
token = credentials.credentials
user_data = auth.verify_token(token)
if user_data is None:
logger.log_message("Token is invalid, access denied to /store.")
raise HTTPException(status_code=403, detail="Not authorized")
user = get_user_by_email(db, email=user_data["sub"])
logger.log_message(f"User {user.email} accessed to /store.")
return templates.TemplateResponse("store.html", {"request": request, "is_superadmin": user.is_superadmin})
# Updated OpenAPI schema to include security definitions
@app.get("/openapi.json", include_in_schema=False)
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = app.openapi()
openapi_schema["components"]["securitySchemes"] = {
"bearerAuth": {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT"
}
}
for path in openapi_schema["paths"]:
for method in openapi_schema["paths"][path]:
openapi_schema["paths"][path][method]["security"] = [
{"bearerAuth": []}]
app.openapi_schema = openapi_schema
return app.openapi_schema