Skip to content

Instantly share code, notes, and snippets.

@nrbnlulu
Created June 4, 2024 10:51
Show Gist options
  • Save nrbnlulu/88d6a91195770db30bab819884c716e0 to your computer and use it in GitHub Desktop.
Save nrbnlulu/88d6a91195770db30bab819884c716e0 to your computer and use it in GitHub Desktop.
Inject parameter for a function using `Concatenate` (for auth purposes)
from collections.abc import Awaitable, Callable
from contextvars import ContextVar
from typing import Concatenate, Protocol
from result.result import Err, Ok, Result
from t5hob_sdk.bases.err import AuthorizationErr
class CheckPermissionCmd:
async def execute(
self, token: str, cmd: type
) -> Result[None, AuthorizationErr]: # pragma: no cover
# TODO: #161 implement permission check
return Ok(None)
class NeedsPermissionCmdProto(Protocol):
permission_cmd: CheckPermissionCmd
def requires_permission[T_self: NeedsPermissionCmdProto, **Ps, T_ret, T_err](
fn: Callable[Concatenate[T_self, Ps], Awaitable[Result[T_ret, T_err]]],
) -> Callable[Concatenate[T_self, str, Ps], Awaitable[Result[T_ret, T_err | AuthorizationErr]]]:
"""Require permission decorator."""
async def wrapper(
self_: T_self, token: str, *args: Ps.args, **kwargs: Ps.kwargs
) -> Result[T_ret, T_err | AuthorizationErr]:
match await CheckPermissionCmd().execute(token, fn.__class__):
case Ok(None):
return await fn(self_, *args, **kwargs)
case Err(err):
return Err(err)
return wrapper
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment