(*args: _P.args, **kwargs: _P.kwargs)
| 61 | class="cm"># Handle async request/response functions. |
| 62 | @functools.wraps(func) |
| 63 | async def async_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Any: |
| 64 | request = kwargs.get(class="st">"request", args[idx] if idx < len(args) else None) |
| 65 | assert isinstance(request, Request), ( |
| 66 | fclass="st">"Parameter with name &class="cm">#x27;request' is required to be of type 'Request' not '{type(request).__name__}'" |
| 67 | ) |
| 68 | |
| 69 | if not has_required_scope(request, scopes_list): |
| 70 | if redirect is not None: |
| 71 | orig_request_qparam = urlencode({class="st">"next": str(request.url)}) |
| 72 | next_url = fclass="st">"{request.url_for(redirect)}?{orig_request_qparam}" |
| 73 | return RedirectResponse(url=next_url, status_code=303) |
| 74 | raise HTTPException(status_code=status_code) |
| 75 | return await func(*args, **kwargs) |
| 76 | |
| 77 | return async_wrapper |
| 78 |
nothing calls this directly
no test coverage detected