from functools import wraps from httpx import AsyncClient, HTTPError from settings import AUTH_URL INTERNAL_AUTH_SERVER = "v2.discours" in AUTH_URL or "testapi.discours" in AUTH_URL async def check_auth(req): token = req.headers.get("Authorization") print(f"[services.auth] checking auth token: {token}") gql = ( {"mutation": "{ getSession { user { id } } }"} if INTERNAL_AUTH_SERVER else {"query": "{ session { user { id } } }"} ) headers = {"Authorization": token, "Content-Type": "application/json"} async with AsyncClient() as client: response = await client.post(AUTH_URL, headers=headers, data=gql) if response.status_code != 200: return False, None r = response.json() print(f"{r}") user_id = ( r.get("data", {}).get("getSession", {}).get("user", {}).get("id", None) if INTERNAL_AUTH_SERVER else r.get("data", {}).get("session", {}).get("user", {}).get("id", None) ) is_authenticated = user_id is not None return is_authenticated, user_id def login_required(f): @wraps(f) async def decorated_function(*args, **kwargs): info = args[1] context = info.context req = context.get("request") is_authenticated, user_id = await check_auth(req) if not is_authenticated: raise Exception("You are not logged in") else: # Добавляем author_id в контекст context["author_id"] = user_id # Если пользователь аутентифицирован, выполняем резолвер return await f(*args, **kwargs) return decorated_function def auth_request(f): @wraps(f) async def decorated_function(*args, **kwargs): req = args[0] is_authenticated, user_id = await check_auth(req) if not is_authenticated: raise HTTPError("please, login first") else: req["author_id"] = user_id return await f(*args, **kwargs) return decorated_function