617 lines
21 KiB
Python
617 lines
21 KiB
Python
import os
|
|
import warnings
|
|
from typing import (
|
|
TYPE_CHECKING,
|
|
Any,
|
|
Callable,
|
|
Dict,
|
|
List,
|
|
Optional,
|
|
Sequence,
|
|
Tuple,
|
|
Type,
|
|
TypeVar,
|
|
Union,
|
|
)
|
|
|
|
from django.http import HttpRequest, HttpResponse
|
|
from django.urls import URLPattern, URLResolver, reverse
|
|
from django.utils.module_loading import import_string
|
|
|
|
from ninja.constants import NOT_SET, NOT_SET_TYPE
|
|
from ninja.errors import (
|
|
ConfigError,
|
|
ValidationError,
|
|
ValidationErrorContext,
|
|
set_default_exc_handlers,
|
|
)
|
|
from ninja.openapi import get_schema
|
|
from ninja.openapi.docs import DocsBase, Swagger
|
|
from ninja.openapi.schema import OpenAPISchema
|
|
from ninja.openapi.urls import get_openapi_urls, get_root_url
|
|
from ninja.parser import Parser
|
|
from ninja.renderers import BaseRenderer, JSONRenderer
|
|
from ninja.router import Router
|
|
from ninja.throttling import BaseThrottle
|
|
from ninja.types import DictStrAny, TCallable
|
|
from ninja.utils import is_debug_server, normalize_path
|
|
|
|
if TYPE_CHECKING:
|
|
from .operation import Operation # pragma: no cover
|
|
|
|
__all__ = ["NinjaAPI"]
|
|
|
|
_E = TypeVar("_E", bound=Exception)
|
|
Exc = Union[_E, Type[_E]]
|
|
ExcHandler = Callable[[HttpRequest, Exc[_E]], HttpResponse]
|
|
|
|
|
|
class NinjaAPI:
|
|
"""
|
|
Ninja API
|
|
"""
|
|
|
|
_registry: List[str] = []
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
title: str = "NinjaAPI",
|
|
version: str = "1.0.0",
|
|
description: str = "",
|
|
openapi_url: Optional[str] = "/openapi.json",
|
|
docs: DocsBase = Swagger(),
|
|
docs_url: Optional[str] = "/docs",
|
|
docs_decorator: Optional[Callable[[TCallable], TCallable]] = None,
|
|
servers: Optional[List[DictStrAny]] = None,
|
|
urls_namespace: Optional[str] = None,
|
|
csrf: bool = False,
|
|
auth: Optional[Union[Sequence[Callable], Callable, NOT_SET_TYPE]] = NOT_SET,
|
|
throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
|
|
renderer: Optional[BaseRenderer] = None,
|
|
parser: Optional[Parser] = None,
|
|
default_router: Optional[Router] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
):
|
|
"""
|
|
Args:
|
|
title: A title for the api.
|
|
description: A description for the api.
|
|
version: The API version.
|
|
urls_namespace: The Django URL namespace for the API. If not provided, the namespace will be ``"api-" + self.version``.
|
|
openapi_url: The relative URL to serve the openAPI spec.
|
|
openapi_extra: Additional attributes for the openAPI spec.
|
|
docs_url: The relative URL to serve the API docs.
|
|
servers: List of target hosts used in openAPI spec.
|
|
csrf: Require a CSRF token for unsafe request types. See <a href="../csrf">CSRF</a> docs.
|
|
auth (Callable | Sequence[Callable] | NOT_SET | None): Authentication class
|
|
renderer: Default response renderer
|
|
parser: Default request parser
|
|
"""
|
|
self.title = title
|
|
self.version = version
|
|
self.description = description
|
|
self.openapi_url = openapi_url
|
|
self.docs = docs
|
|
self.docs_url = docs_url
|
|
self.docs_decorator = docs_decorator
|
|
self.servers = servers or []
|
|
self.urls_namespace = urls_namespace or f"api-{self.version}"
|
|
self.csrf = csrf # TODO: Check if used or at least throw Deprecation warning
|
|
if self.csrf:
|
|
warnings.warn(
|
|
"csrf argument is deprecated, auth is handling csrf automatically now",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
self.renderer = renderer or JSONRenderer()
|
|
self.parser = parser or Parser()
|
|
self.openapi_extra = openapi_extra or {}
|
|
|
|
self._exception_handlers: Dict[Exc, ExcHandler] = {}
|
|
self.set_default_exception_handlers()
|
|
|
|
self.auth: Optional[Union[Sequence[Callable], NOT_SET_TYPE]]
|
|
|
|
if callable(auth):
|
|
self.auth = [auth]
|
|
else:
|
|
self.auth = auth
|
|
|
|
self.throttle = throttle
|
|
|
|
self._routers: List[Tuple[str, Router]] = []
|
|
self.default_router = default_router or Router()
|
|
self.add_router("", self.default_router)
|
|
|
|
def get(
|
|
self,
|
|
path: str,
|
|
*,
|
|
auth: Any = NOT_SET,
|
|
throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
|
|
response: Any = NOT_SET,
|
|
operation_id: Optional[str] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
tags: Optional[List[str]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
by_alias: Optional[bool] = None,
|
|
exclude_unset: Optional[bool] = None,
|
|
exclude_defaults: Optional[bool] = None,
|
|
exclude_none: Optional[bool] = None,
|
|
url_name: Optional[str] = None,
|
|
include_in_schema: bool = True,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
) -> Callable[[TCallable], TCallable]:
|
|
"""
|
|
`GET` operation. See <a href="../operations-parameters">operations
|
|
parameters</a> reference.
|
|
"""
|
|
return self.default_router.get(
|
|
path,
|
|
auth=auth is NOT_SET and self.auth or auth,
|
|
throttle=throttle is NOT_SET and self.throttle or throttle,
|
|
response=response,
|
|
operation_id=operation_id,
|
|
summary=summary,
|
|
description=description,
|
|
tags=tags,
|
|
deprecated=deprecated,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
exclude_defaults=exclude_defaults,
|
|
exclude_none=exclude_none,
|
|
url_name=url_name,
|
|
include_in_schema=include_in_schema,
|
|
openapi_extra=openapi_extra,
|
|
)
|
|
|
|
def post(
|
|
self,
|
|
path: str,
|
|
*,
|
|
auth: Any = NOT_SET,
|
|
throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
|
|
response: Any = NOT_SET,
|
|
operation_id: Optional[str] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
tags: Optional[List[str]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
by_alias: Optional[bool] = None,
|
|
exclude_unset: Optional[bool] = None,
|
|
exclude_defaults: Optional[bool] = None,
|
|
exclude_none: Optional[bool] = None,
|
|
url_name: Optional[str] = None,
|
|
include_in_schema: bool = True,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
) -> Callable[[TCallable], TCallable]:
|
|
"""
|
|
`POST` operation. See <a href="../operations-parameters">operations
|
|
parameters</a> reference.
|
|
"""
|
|
return self.default_router.post(
|
|
path,
|
|
auth=auth is NOT_SET and self.auth or auth,
|
|
throttle=throttle is NOT_SET and self.throttle or throttle,
|
|
response=response,
|
|
operation_id=operation_id,
|
|
summary=summary,
|
|
description=description,
|
|
tags=tags,
|
|
deprecated=deprecated,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
exclude_defaults=exclude_defaults,
|
|
exclude_none=exclude_none,
|
|
url_name=url_name,
|
|
include_in_schema=include_in_schema,
|
|
openapi_extra=openapi_extra,
|
|
)
|
|
|
|
def delete(
|
|
self,
|
|
path: str,
|
|
*,
|
|
auth: Any = NOT_SET,
|
|
throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
|
|
response: Any = NOT_SET,
|
|
operation_id: Optional[str] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
tags: Optional[List[str]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
by_alias: Optional[bool] = None,
|
|
exclude_unset: Optional[bool] = None,
|
|
exclude_defaults: Optional[bool] = None,
|
|
exclude_none: Optional[bool] = None,
|
|
url_name: Optional[str] = None,
|
|
include_in_schema: bool = True,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
) -> Callable[[TCallable], TCallable]:
|
|
"""
|
|
`DELETE` operation. See <a href="../operations-parameters">operations
|
|
parameters</a> reference.
|
|
"""
|
|
return self.default_router.delete(
|
|
path,
|
|
auth=auth is NOT_SET and self.auth or auth,
|
|
throttle=throttle is NOT_SET and self.throttle or throttle,
|
|
response=response,
|
|
operation_id=operation_id,
|
|
summary=summary,
|
|
description=description,
|
|
tags=tags,
|
|
deprecated=deprecated,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
exclude_defaults=exclude_defaults,
|
|
exclude_none=exclude_none,
|
|
url_name=url_name,
|
|
include_in_schema=include_in_schema,
|
|
openapi_extra=openapi_extra,
|
|
)
|
|
|
|
def patch(
|
|
self,
|
|
path: str,
|
|
*,
|
|
auth: Any = NOT_SET,
|
|
throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
|
|
response: Any = NOT_SET,
|
|
operation_id: Optional[str] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
tags: Optional[List[str]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
by_alias: Optional[bool] = None,
|
|
exclude_unset: Optional[bool] = None,
|
|
exclude_defaults: Optional[bool] = None,
|
|
exclude_none: Optional[bool] = None,
|
|
url_name: Optional[str] = None,
|
|
include_in_schema: bool = True,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
) -> Callable[[TCallable], TCallable]:
|
|
"""
|
|
`PATCH` operation. See <a href="../operations-parameters">operations
|
|
parameters</a> reference.
|
|
"""
|
|
return self.default_router.patch(
|
|
path,
|
|
auth=auth is NOT_SET and self.auth or auth,
|
|
throttle=throttle is NOT_SET and self.throttle or throttle,
|
|
response=response,
|
|
operation_id=operation_id,
|
|
summary=summary,
|
|
description=description,
|
|
tags=tags,
|
|
deprecated=deprecated,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
exclude_defaults=exclude_defaults,
|
|
exclude_none=exclude_none,
|
|
url_name=url_name,
|
|
include_in_schema=include_in_schema,
|
|
openapi_extra=openapi_extra,
|
|
)
|
|
|
|
def put(
|
|
self,
|
|
path: str,
|
|
*,
|
|
auth: Any = NOT_SET,
|
|
throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
|
|
response: Any = NOT_SET,
|
|
operation_id: Optional[str] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
tags: Optional[List[str]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
by_alias: Optional[bool] = None,
|
|
exclude_unset: Optional[bool] = None,
|
|
exclude_defaults: Optional[bool] = None,
|
|
exclude_none: Optional[bool] = None,
|
|
url_name: Optional[str] = None,
|
|
include_in_schema: bool = True,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
) -> Callable[[TCallable], TCallable]:
|
|
"""
|
|
`PUT` operation. See <a href="../operations-parameters">operations
|
|
parameters</a> reference.
|
|
"""
|
|
return self.default_router.put(
|
|
path,
|
|
auth=auth is NOT_SET and self.auth or auth,
|
|
throttle=throttle is NOT_SET and self.throttle or throttle,
|
|
response=response,
|
|
operation_id=operation_id,
|
|
summary=summary,
|
|
description=description,
|
|
tags=tags,
|
|
deprecated=deprecated,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
exclude_defaults=exclude_defaults,
|
|
exclude_none=exclude_none,
|
|
url_name=url_name,
|
|
include_in_schema=include_in_schema,
|
|
openapi_extra=openapi_extra,
|
|
)
|
|
|
|
def api_operation(
|
|
self,
|
|
methods: List[str],
|
|
path: str,
|
|
*,
|
|
auth: Any = NOT_SET,
|
|
throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
|
|
response: Any = NOT_SET,
|
|
operation_id: Optional[str] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
tags: Optional[List[str]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
by_alias: Optional[bool] = None,
|
|
exclude_unset: Optional[bool] = None,
|
|
exclude_defaults: Optional[bool] = None,
|
|
exclude_none: Optional[bool] = None,
|
|
url_name: Optional[str] = None,
|
|
include_in_schema: bool = True,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
) -> Callable[[TCallable], TCallable]:
|
|
return self.default_router.api_operation(
|
|
methods,
|
|
path,
|
|
auth=auth is NOT_SET and self.auth or auth,
|
|
throttle=throttle is NOT_SET and self.throttle or throttle,
|
|
response=response,
|
|
operation_id=operation_id,
|
|
summary=summary,
|
|
description=description,
|
|
tags=tags,
|
|
deprecated=deprecated,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
exclude_defaults=exclude_defaults,
|
|
exclude_none=exclude_none,
|
|
url_name=url_name,
|
|
include_in_schema=include_in_schema,
|
|
openapi_extra=openapi_extra,
|
|
)
|
|
|
|
def add_router(
|
|
self,
|
|
prefix: str,
|
|
router: Union[Router, str],
|
|
*,
|
|
auth: Any = NOT_SET,
|
|
throttle: Union[BaseThrottle, List[BaseThrottle], NOT_SET_TYPE] = NOT_SET,
|
|
tags: Optional[List[str]] = None,
|
|
parent_router: Optional[Router] = None,
|
|
) -> None:
|
|
if isinstance(router, str):
|
|
router = import_string(router)
|
|
assert isinstance(router, Router)
|
|
|
|
if auth is not NOT_SET:
|
|
router.auth = auth
|
|
|
|
if throttle is not NOT_SET:
|
|
router.throttle = throttle
|
|
|
|
if tags is not None:
|
|
router.tags = tags
|
|
|
|
if parent_router:
|
|
parent_prefix = next(
|
|
(path for path, r in self._routers if r is parent_router), None
|
|
) # pragma: no cover
|
|
assert parent_prefix is not None
|
|
prefix = normalize_path("/".join((parent_prefix, prefix))).lstrip("/")
|
|
|
|
self._routers.extend(router.build_routers(prefix))
|
|
router.set_api_instance(self, parent_router)
|
|
|
|
@property
|
|
def urls(self) -> Tuple[List[Union[URLResolver, URLPattern]], str, str]:
|
|
"""
|
|
str: URL configuration
|
|
|
|
Returns:
|
|
|
|
Django URL configuration
|
|
"""
|
|
self._validate()
|
|
return (
|
|
self._get_urls(),
|
|
"ninja",
|
|
self.urls_namespace.split(":")[-1],
|
|
# ^ if api included into nested urls, we only care about last bit here
|
|
)
|
|
|
|
def _get_urls(self) -> List[Union[URLResolver, URLPattern]]:
|
|
result = get_openapi_urls(self)
|
|
|
|
for prefix, router in self._routers:
|
|
result.extend(router.urls_paths(prefix))
|
|
|
|
result.append(get_root_url(self))
|
|
return result
|
|
|
|
def get_root_path(self, path_params: DictStrAny) -> str:
|
|
name = f"{self.urls_namespace}:api-root"
|
|
return reverse(name, kwargs=path_params)
|
|
|
|
def create_response(
|
|
self,
|
|
request: HttpRequest,
|
|
data: Any,
|
|
*,
|
|
status: Optional[int] = None,
|
|
temporal_response: Optional[HttpResponse] = None,
|
|
) -> HttpResponse:
|
|
if temporal_response:
|
|
status = temporal_response.status_code
|
|
assert status
|
|
|
|
content = self.renderer.render(request, data, response_status=status)
|
|
|
|
if temporal_response:
|
|
response = temporal_response
|
|
response.content = content
|
|
else:
|
|
response = HttpResponse(
|
|
content, status=status, content_type=self.get_content_type()
|
|
)
|
|
|
|
return response
|
|
|
|
def create_temporal_response(self, request: HttpRequest) -> HttpResponse:
|
|
return HttpResponse("", content_type=self.get_content_type())
|
|
|
|
def get_content_type(self) -> str:
|
|
return f"{self.renderer.media_type}; charset={self.renderer.charset}"
|
|
|
|
def get_openapi_schema(
|
|
self,
|
|
*,
|
|
path_prefix: Optional[str] = None,
|
|
path_params: Optional[DictStrAny] = None,
|
|
) -> OpenAPISchema:
|
|
if path_prefix is None:
|
|
path_prefix = self.get_root_path(path_params or {})
|
|
return get_schema(api=self, path_prefix=path_prefix)
|
|
|
|
def get_openapi_operation_id(self, operation: "Operation") -> str:
|
|
name = operation.view_func.__name__
|
|
module = operation.view_func.__module__
|
|
return (module + "_" + name).replace(".", "_")
|
|
|
|
def get_operation_url_name(self, operation: "Operation", router: Router) -> str:
|
|
"""
|
|
Get the default URL name to use for an operation if it wasn't
|
|
explicitly provided.
|
|
"""
|
|
return operation.view_func.__name__
|
|
|
|
def add_exception_handler(
|
|
self, exc_class: Type[_E], handler: ExcHandler[_E]
|
|
) -> None:
|
|
assert issubclass(exc_class, Exception)
|
|
self._exception_handlers[exc_class] = handler
|
|
|
|
def exception_handler(
|
|
self, exc_class: Type[Exception]
|
|
) -> Callable[[TCallable], TCallable]:
|
|
def decorator(func: TCallable) -> TCallable:
|
|
self.add_exception_handler(exc_class, func)
|
|
return func
|
|
|
|
return decorator
|
|
|
|
def set_default_exception_handlers(self) -> None:
|
|
set_default_exc_handlers(self)
|
|
|
|
def on_exception(self, request: HttpRequest, exc: Exc[_E]) -> HttpResponse:
|
|
handler = self._lookup_exception_handler(exc)
|
|
if handler is None:
|
|
raise exc
|
|
return handler(request, exc)
|
|
|
|
def validation_error_from_error_contexts(
|
|
self, error_contexts: List[ValidationErrorContext]
|
|
) -> ValidationError:
|
|
errors: List[Dict[str, Any]] = []
|
|
for context in error_contexts:
|
|
model = context.model
|
|
e = context.pydantic_validation_error
|
|
for i in e.errors(include_url=False):
|
|
i["loc"] = (
|
|
model.__ninja_param_source__,
|
|
) + model.__ninja_flatten_map_reverse__.get(i["loc"], i["loc"])
|
|
# removing pydantic hints
|
|
del i["input"] # type: ignore
|
|
if (
|
|
"ctx" in i
|
|
and "error" in i["ctx"]
|
|
and isinstance(i["ctx"]["error"], Exception)
|
|
):
|
|
i["ctx"]["error"] = str(i["ctx"]["error"])
|
|
errors.append(dict(i))
|
|
return ValidationError(errors)
|
|
|
|
def _lookup_exception_handler(self, exc: Exc[_E]) -> Optional[ExcHandler[_E]]:
|
|
for cls in type(exc).__mro__:
|
|
if cls in self._exception_handlers:
|
|
return self._exception_handlers[cls]
|
|
|
|
return None
|
|
|
|
def _validate(self) -> None:
|
|
# urls namespacing validation
|
|
skip_registry = os.environ.get("NINJA_SKIP_REGISTRY", False)
|
|
if (
|
|
not skip_registry
|
|
and self.urls_namespace in NinjaAPI._registry
|
|
and not debug_server_url_reimport()
|
|
):
|
|
msg = f"""
|
|
Looks like you created multiple NinjaAPIs or TestClients
|
|
To let ninja distinguish them you need to set either unique version or urls_namespace
|
|
- NinjaAPI(..., version='2.0.0')
|
|
- NinjaAPI(..., urls_namespace='otherapi')
|
|
Already registered: {NinjaAPI._registry}
|
|
"""
|
|
raise ConfigError(msg.strip())
|
|
NinjaAPI._registry.append(self.urls_namespace)
|
|
|
|
|
|
_imported_while_running_in_debug_server = is_debug_server()
|
|
|
|
|
|
def debug_server_url_reimport() -> bool:
|
|
"""
|
|
Detect reimport of URL module to allow error to propagate to developer.
|
|
|
|
When Django loads urls it uses: ``django.urls.resolvers.urlconf_module()``
|
|
|
|
```Python
|
|
@cached_property
|
|
def urlconf_module(self):
|
|
if isinstance(self.urlconf_name, str):
|
|
return import_module(self.urlconf_name)
|
|
else:
|
|
return self.urlconf_name
|
|
```
|
|
|
|
This uses ``@cached_property`` to generally only import once. But if the
|
|
import throws an error when using the development server, the following
|
|
code in ``django.utils.autoreload.BaseReloader.run()`` is used:
|
|
|
|
```Python
|
|
# Prevent a race condition where URL modules aren't loaded when the
|
|
# reloader starts by accessing the urlconf_module property.
|
|
try:
|
|
get_resolver().urlconf_module
|
|
except Exception:
|
|
# Loading the urlconf can result in errors during development.
|
|
# If this occurs then swallow the error and continue.
|
|
pass
|
|
```
|
|
|
|
This means the (likely) developer error that caused the Exception is
|
|
initially ignored. This is not generally a problem since the error will
|
|
usually be exercised again, and reported at that time. But Ninja has
|
|
various code which guards against errors where items that cannot be reused,
|
|
are attempted to be reused. This results in Ninja throwing a false error,
|
|
and hiding the true error from the developer when running under the
|
|
development server.
|
|
|
|
Returns:
|
|
|
|
True if this module was originally imported during Django dev-server
|
|
init but the caller is not being running during Django dev-server init.
|
|
"""
|
|
return _imported_while_running_in_debug_server and not is_debug_server()
|