Function decorator

starlette/authentication.py:32–96

(
        func: Callable[_P, Any],
    )

    scopes_list = [scopes] if isinstance(scopes, str) else list(scopes)

    def decorator(
        func: Callable[_P, Any],
    ) -> Callable[_P, Any]:
        sig = inspect.signature(func)
        for idx, parameter in enumerate(sig.parameters.values()):
            if parameter.name == class="st">"request" or parameter.name == class="st">"websocket":
                type_ = parameter.name
                break
        else:
            raise Exception(f&class="cm">#x27;No class="st">"request" or class="st">"websocket" argument on function class="st">"{func}"')

        if type_ == class="st">"websocket":
            class="cm"># Handle websocket functions. (Always async)
            @functools.wraps(func)
            async def websocket_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
                websocket = kwargs.get(class="st">"websocket", args[idx] if idx < len(args) else None)
                assert isinstance(websocket, WebSocket), (
                    class="st">"Parameter with name &class="cm">#x27;websocket' is required to be of type 'WebSocket'"
                    fclass="st">" not &class="cm">#x27;{type(websocket).__name__}'"
                )

                if not has_required_scope(websocket, scopes_list):
                    await websocket.close()
                else:
                    await func(*args, **kwargs)

            return websocket_wrapper

        elif is_async_callable(func):
            class="cm"># Handle async request/response functions.
            @functools.wraps(func)
            async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any:
                request = kwargs.get(class="st">"request", args[idx] if idx < len(args) else None)
                assert isinstance(request, Request), (
                    fclass="st">"Parameter with name &class="cm">#x27;request' is required to be of type 'Request' not '{type(request).__name__}'"
                )

                if not has_required_scope(request, scopes_list):
                    if redirect is not None:
                        orig_request_qparam = urlencode({class="st">"next": str(request.url)})
                        next_url = fclass="st">"{request.url_for(redirect)}?{orig_request_qparam}"
                        return RedirectResponse(url=next_url, status_code=303)
                    raise HTTPException(status_code=status_code)
                return await func(*args, **kwargs)

            return async_wrapper

        else:
            class="cm"># Handle sync request/response functions.
            @functools.wraps(func)
            def sync_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any:
                request = kwargs.get(class="st">"request", args[idx] if idx < len(args) else None)
                assert isinstance(request, Request), (
                    fclass="st">"Parameter with name &class="cm">#x27;request' is required to be of type 'Request' not '{type(request).__name__}'"
                )

                if not has_required_scope(request, scopes_list):
                    if redirect is not None: