/
FlysAt
/
DataFoundry
Обзор
Документация
Войти
/
FlysAt
/
DataFoundry
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
backend/src/chat/auth.py
35 строк
1 KB
AlexanderShmygol
add chat
04 май 2026, 21:47
04 май 2026, 21:47
af51979
Код
Авторство
О чём код?
from fastapi import HTTPException, WebSocket, status from sqlalchemy.ext.asyncio import AsyncSession from src.api.repositories.user import UserRepository from src.core.security import decode_token from src.models.user import User async def authenticate_websocket(websocket: WebSocket, db: AsyncSession) -> User: token = websocket.query_params.get("token") if not token: authorization = websocket.headers.get("authorization") if authorization and authorization.lower().startswith("bearer "): token = authorization[7:] if not token: await websocket.close(code=status.WS_1008_POLICY_VIOLATION) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing token") try: payload = decode_token(token) if payload.get("type") != "access": raise ValueError("Invalid token type") user_id = payload.get("sub") if user_id is None: raise ValueError("Invalid token payload") except ValueError as exc: await websocket.close(code=status.WS_1008_POLICY_VIOLATION) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") from exc user = await UserRepository(db).get_by_id(int(user_id)) if not user: await websocket.close(code=status.WS_1008_POLICY_VIOLATION) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found") return user