74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
import inspect
|
|
from typing import Any, Callable, Optional, Type
|
|
|
|
from django.conf import settings
|
|
from django.http import HttpRequest, HttpResponseForbidden
|
|
from django.middleware.csrf import CsrfViewMiddleware
|
|
|
|
__all__ = [
|
|
"check_csrf",
|
|
"is_debug_server",
|
|
"normalize_path",
|
|
"contribute_operation_callback",
|
|
]
|
|
|
|
|
|
def replace_path_param_notation(path: str) -> str:
|
|
return path.replace("{", "<").replace("}", ">")
|
|
|
|
|
|
def normalize_path(path: str) -> str:
|
|
while "//" in path:
|
|
path = path.replace("//", "/")
|
|
return path
|
|
|
|
|
|
def _no_view() -> None:
|
|
pass # pragma: no cover
|
|
|
|
|
|
def check_csrf(
|
|
request: HttpRequest, callback: Callable = _no_view
|
|
) -> Optional[HttpResponseForbidden]:
|
|
mware = CsrfViewMiddleware(lambda x: HttpResponseForbidden()) # pragma: no cover
|
|
request.csrf_processing_done = False # type: ignore
|
|
mware.process_request(request)
|
|
return mware.process_view(request, callback, (), {})
|
|
|
|
|
|
def is_debug_server() -> bool:
|
|
"""Check if running under the Django Debug Server"""
|
|
return settings.DEBUG and any(
|
|
s.filename.endswith("runserver.py") and s.function == "run"
|
|
for s in inspect.stack(0)[1:]
|
|
)
|
|
|
|
|
|
def is_async_callable(f: Callable[..., Any]) -> bool:
|
|
return inspect.iscoroutinefunction(f) or inspect.iscoroutinefunction(
|
|
getattr(f, "__call__", None)
|
|
)
|
|
|
|
|
|
def is_optional_type(t: Type[Any]) -> bool:
|
|
try:
|
|
return type(None) in t.__args__
|
|
except AttributeError:
|
|
return False
|
|
|
|
|
|
def contribute_operation_callback(
|
|
func: Callable[..., Any], callback: Callable[..., Any]
|
|
) -> None:
|
|
if not hasattr(func, "_ninja_contribute_to_operation"):
|
|
func._ninja_contribute_to_operation = [] # type: ignore
|
|
func._ninja_contribute_to_operation.append(callback) # type: ignore
|
|
|
|
|
|
def contribute_operation_args(
|
|
func: Callable[..., Any], arg_name: str, arg_type: Type, arg_source: Any
|
|
) -> None:
|
|
if not hasattr(func, "_ninja_contribute_args"):
|
|
func._ninja_contribute_args = [] # type: ignore
|
|
func._ninja_contribute_args.append((arg_name, arg_type, arg_source)) # type: ignore
|