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 == "request" or parameter.name == "websocket":
                type_ = parameter.name
                break
        else:
            raise Exception(f'No "request" or "websocket" argument on function "{func}"')

        if type_ == "websocket":
            # Handle websocket functions. (Always async)
            @functools.wraps(func)
            async def websocket_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
                websocket = kwargs.get("websocket", args[idx] if idx < len(args) else None)
                assert isinstance(websocket, WebSocket), (
                    "Parameter with name 'websocket' is required to be of type 'WebSocket'"
                    f" not '{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):
            # Handle async request/response functions.
            @functools.wraps(func)
            async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any:
                request = kwargs.get("request", args[idx] if idx < len(args) else None)
                assert isinstance(request, Request), (
                    f"Parameter with name '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({"next": str(request.url)})
                        next_url = f"{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:
            # Handle sync request/response functions.
            @functools.wraps(func)
            def sync_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any:
                request = kwargs.get("request", args[idx] if idx < len(args) else None)
                assert isinstance(request, Request), (
                    f"Parameter with name '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: