Spaces:
Running
Running
File size: 2,981 Bytes
c7426d8 db83efb c7426d8 db83efb c7426d8 db83efb 0254a24 c7426d8 f06d4d5 db83efb c7426d8 db83efb c7426d8 |
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 |
"""Module to handle the main FastAPI application and its endpoints."""
import logging
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from jose import jwt
from router import auth, content, mail
from starlette.middleware.base import BaseHTTPMiddleware
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
class SessionMiddleware(BaseHTTPMiddleware):
"""
Middleware to manage session data using JWT (JSON Web Tokens) stored in cookies.
This middleware intercepts incoming HTTP requests to extract session data from a
"session_token" cookie. If the cookie exists and contains a valid JWT, the session
data is decoded and attached to the request's state. If the cookie is missing or
invalid, an empty session is initialized.
After processing the request, the middleware encodes the updated session data into
a new JWT and sets it as a "session_token" cookie in the response.
Attributes:
SECRET_KEY (str): The secret key used to encode and decode the JWT.
ALGORITHM (str): The algorithm used for encoding and decoding the JWT.
Methods:
dispatch(request: Request, call_next): Intercepts the request to manage session
data and modifies the response to include the updated session token.
Cookie Parameters:
session_token (str): A JWT containing session data. This cookie is HTTP-only
and has a maximum age of 3600 seconds (1 hour).
Raises:
jwt.JWTError: If the session token cannot be decoded due to invalid signature
or other issues.
"""
async def dispatch(self, request: Request, call_next):
session_token = request.cookies.get("session_token")
if session_token:
try:
session_data = jwt.decode(session_token, SECRET_KEY, algorithms=[ALGORITHM])
except jwt.JWTError:
session_data = {}
else:
session_data = {}
request.state.session = session_data
response = await call_next(request)
session_token = jwt.encode(request.state.session, SECRET_KEY, algorithm=ALGORITHM)
response.set_cookie(
key="session_token",
value=session_token,
httponly=True,
max_age=3600
)
return response
logging.basicConfig(
format='%(asctime)s - %(levelname)s - %(funcName)s - %(message)s')
logging.getLogger().setLevel(logging.INFO)
app = FastAPI(docs_url="/")
app.include_router(content.router)
app.include_router(mail.router)
app.include_router(auth.router)
origins = [
"*"
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials = True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_middleware(SessionMiddleware)
@app.get("/_health")
def health():
"""
Returns the health status of the application.
:return: A string "OK" indicating the health status.
"""
return "OK"
|