from functools import wraps from starlette.responses import JSONResponse import aiohttp AUTH_URL = 'https://auth.discours.io' async def check_auth(req): token = req.headers.get("Authorization") headers = {"Authorization": token, "Content-Type": "application/json"} print(f"[services.auth] checking auth token: {token}") gql = { "query": "query GetUserId { session { user { id } } }", "operationName": "GetUserId", "variables": None, } async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30.0)) as session: async with session.post(AUTH_URL, headers=headers, json=gql) as response: print(f"[services.auth] {AUTH_URL} response: {response.status}") if response.status != 200: return False, None r = await response.json() if r: user_id = r.get("data", {}).get("session", {}).get("user", {}).get("id", None) is_authenticated = user_id is not None return is_authenticated, user_id return False, None def login_required(f): @wraps(f) async def decorated_function(request, *args, **kwargs): is_authenticated, user_id = await check_auth(request) if not is_authenticated: return JSONResponse({'error': 'Unauthorized'}, status_code=401) # Make user_id available to the route handler, if needed request.state.user_id = user_id return await f(request, *args, **kwargs) return decorated_function