from typing import Callable from aiohttp import web from server.core.exceptions import Unauthorized from server.core.schemas import IdSchema class Helper: @staticmethod def args_decompose(*args, **kwargs): def decorator(func): async def wrapper(*args, **kwargs): request = args[0] kwargs = request.match_info if "id" in kwargs: kwargs = IdSchema().load(kwargs) kwargs["id"] = int(kwargs["id"]) return await func(request, **dict(kwargs)) return wrapper return decorator @staticmethod def auth_with_user_id(handler: Callable) -> Callable: """Authentication performed by extracting user_id from header . Use for handlers where user_id is required. Args: handler: handler function to wrap Returns: wrapped handler Raises: Unauthorized: If user_id is missing. """ async def wrapper(*args, **kwargs) -> web.Response: request = args[0] user_id = request.headers.get("X-User-Id") if not user_id: raise Unauthorized("User ID is not provided.") request["user_id"] = user_id return await handler(request, **dict(kwargs)) return wrapper @staticmethod def check_request_url(schema): def decorator(func): async def wrapper(*args, **kwargs): request = args[0] kwargs = schema().load(kwargs) return await func(request, **dict(kwargs)) return wrapper return decorator