29 lines
1.1 KiB
Python
29 lines
1.1 KiB
Python
|
from typing import Any
|
||
|
|
|
||
|
|
from app.core.jwt import decode_token
|
||
|
|
from app.db import SessionLocal
|
||
|
|
from app.models import User
|
||
|
|
|
||
|
|
|
||
|
|
def get_context_value(request: Any, data: Any = None) -> dict[str, Any]:
|
||
|
|
# создать сессию SQLAlchemy (`SessionLocal()`);
|
||
|
|
db = SessionLocal()
|
||
|
|
user = None
|
||
|
|
|
||
|
|
# извлечь заголовок `Authorization`;
|
||
|
|
authorization = request.headers.get("authorization")
|
||
|
|
if authorization and authorization.lower().startswith("bearer "):
|
||
|
|
# если токен передан и валиден – найти пользователя и положить его в контекст;
|
||
|
|
try:
|
||
|
|
payload = decode_token(authorization.split(" ", 1)[1])
|
||
|
|
sub = payload.get("sub")
|
||
|
|
if isinstance(sub, str) and sub.isdigit():
|
||
|
|
candidate = db.get(User, int(sub))
|
||
|
|
if candidate is not None and candidate.is_active:
|
||
|
|
user = candidate
|
||
|
|
except ValueError:
|
||
|
|
user = None
|
||
|
|
|
||
|
|
# вернуть словарь `{"request": ..., "db": ..., "user": ...}`.
|
||
|
|
return {"request": request, "db": db, "user": user}
|