import inspect
import logging
from rest_framework import serializers
from .. import openapi
from ..utils import force_real_str, get_field_default, get_object_classes, is_list_view
#: Sentinel value that inspectors must return to signal that they do not know how to
# handle an object
NotHandled = object()
logger = logging.getLogger(__name__)
def is_callable_method(cls_or_instance, method_name):
method = getattr(cls_or_instance, method_name)
if inspect.ismethod(method) and getattr(method, "__self__", None):
# bound classmethod or instance method
return method, True
from inspect import getattr_static
return method, isinstance(
getattr_static(cls_or_instance, method_name, None), staticmethod
)
def call_view_method(view, method_name, fallback_attr=None, default=None):
"""Call a view method which might throw an exception. If an exception is thrown, log
an informative error message and return the value of fallback_attr, or default if
not present. The method must be callable without any arguments except cls or self.
:param view: view class or instance; if a class is passed, instance methods won't be
called
:type view: rest_framework.views.APIView or type[rest_framework.views.APIView]
:param str method_name: name of a method on the view
:param str fallback_attr: name of an attribute on the view to fall back on, if
calling the method fails
:param default: default value if all else fails
:return: view method's return value, or value of view's fallback_attr, or default
:rtype: any or None
"""
if hasattr(view, method_name):
try:
view_method, is_callabale = is_callable_method(view, method_name)
if is_callabale:
return view_method()
except Exception: # pragma: no cover
logger.warning(
"view's %s.%s raised exception during schema generation; use "
"`getattr(self, 'swagger_fake_view', False)` to detect and "
"short-circuit this",
type(view).__name__,
method_name,
exc_info=True,
)
if fallback_attr and hasattr(view, fallback_attr):
return getattr(view, fallback_attr)
return default
class BaseInspector:
def __init__(self, view, path, method, components, request):
"""
:param rest_framework.views.APIView view: the view associated with this endpoint
:param str path: the path component of the operation URL
:param str method: the http method of the operation
:param openapi.ReferenceResolver components: referenceable components
:param rest_framework.request.Request request: the request made against the
schema view; can be None
"""
self.view = view
self.path = path
self.method = method
self.components = components
self.request = request
[docs]
def process_result(self, result, method_name, obj, **kwargs):
"""After an inspector handles an object (i.e. returns a value other than
:data:`.NotHandled`), all inspectors that were probed get the chance to alter
the result, in reverse order. The inspector that handled the object is the first
to receive a ``process_result`` call with the object it just returned.
This behavior is similar to the Django request/response middleware processing.
If this inspector has no post-processing to do, it should just ``return result``
(the default implementation).
:param result: the return value of the winning inspector, or ``None`` if no
inspector handled the object
:param str method_name: name of the method that was called on the inspector
:param obj: first argument passed to inspector method
:param kwargs: additional arguments passed to inspector method
:return:
"""
return result
[docs]
def probe_inspectors(self, inspectors, method_name, obj, initkwargs=None, **kwargs):
"""Probe a list of inspectors with a given object. The first inspector in the
list to return a value that is not :data:`.NotHandled` wins.
:param list[type[BaseInspector]] inspectors: list of inspectors to probe
:param str method_name: name of the target method on the inspector
:param obj: first argument to inspector method
:param dict initkwargs: extra kwargs for instantiating inspector class
:param kwargs: additional arguments to inspector method
:return: the return value of the winning inspector, or ``None`` if no inspector
handled the object
"""
initkwargs = initkwargs or {}
tried_inspectors = []
for inspector in inspectors:
assert inspect.isclass(inspector), (
"inspector must be a class, not an object"
)
assert issubclass(inspector, BaseInspector), (
"inspectors must subclass BaseInspector"
)
inspector = inspector(
self.view,
self.path,
self.method,
self.components,
self.request,
**initkwargs,
)
tried_inspectors.append(inspector)
method = getattr(inspector, method_name, None)
if method is None:
continue
result = method(obj, **kwargs)
if result is not NotHandled:
break
else: # pragma: no cover
logger.warning(
"%s ignored because no inspector in %s handled it (operation: %s)",
obj,
inspectors,
method_name,
)
result = None
for inspector in reversed(tried_inspectors):
result = inspector.process_result(result, method_name, obj, **kwargs)
return result
[docs]
def get_renderer_classes(self):
"""Get the renderer classes of this view by calling `get_renderers`.
:return: renderer classes
:rtype: list[type[rest_framework.renderers.BaseRenderer]]
"""
return get_object_classes(
call_view_method(self.view, "get_renderers", "renderer_classes", [])
)
[docs]
def get_parser_classes(self):
"""Get the parser classes of this view by calling `get_parsers`.
:return: parser classes
:rtype: list[type[rest_framework.parsers.BaseParser]]
"""
return get_object_classes(
call_view_method(self.view, "get_parsers", "parser_classes", [])
)
class PaginatorInspector(BaseInspector):
"""Base inspector for paginators.
Responsible for determining extra query parameters and response structure added by
given paginators.
"""
[docs]
def get_paginator_parameters(self, paginator):
"""Get the pagination parameters for a single paginator **instance**.
Should return :data:`.NotHandled` if this inspector does not know how to handle
the given `paginator`.
:param BasePagination paginator: the paginator
:rtype: list[openapi.Parameter]
"""
return NotHandled
[docs]
def get_paginated_response(self, paginator, response_schema):
"""Add appropriate paging fields to a response :class:`.Schema`.
Should return :data:`.NotHandled` if this inspector does not know how to handle
the given `paginator`.
:param BasePagination paginator: the paginator
:param openapi.Schema response_schema: the response schema that must be paged.
:rtype: openapi.Schema
"""
return NotHandled
class FilterInspector(BaseInspector):
"""Base inspector for filter backends.
Responsible for determining extra query parameters added by given filter backends.
"""
[docs]
def get_filter_parameters(self, filter_backend):
"""Get the filter parameters for a single filter backend **instance**.
Should return :data:`.NotHandled` if this inspector does not know how to handle
the given `filter_backend`.
:param BaseFilterBackend filter_backend: the filter backend
:rtype: list[openapi.Parameter]
"""
return NotHandled
class FieldInspector(BaseInspector):
"""Base inspector for serializers and serializer fields."""
def __init__(self, view, path, method, components, request, field_inspectors):
super(FieldInspector, self).__init__(view, path, method, components, request)
self.field_inspectors = field_inspectors
[docs]
def add_manual_fields(self, serializer_or_field, schema):
"""Set fields from the ``swagger_schema_fields`` attribute on the Meta class.
This method is called only for serializers or fields that are converted into
``openapi.Schema`` objects.
:param serializer_or_field: serializer or field instance
:param openapi.Schema schema: the schema object to be modified in-place
"""
meta = getattr(serializer_or_field, "Meta", None)
swagger_schema_fields = getattr(meta, "swagger_schema_fields", {})
if swagger_schema_fields:
for attr, val in swagger_schema_fields.items():
setattr(schema, attr, val)
[docs]
def field_to_swagger_object(
self, field, swagger_object_type, use_references, **kwargs
):
"""Convert a drf Serializer or Field instance into a Swagger object.
Should return :data:`.NotHandled` if this inspector does not know how to handle
the given `field`.
:param rest_framework.serializers.Field field: the source field
:param type[openapi.SwaggerDict] swagger_object_type: should be one of Schema,
Parameter, Items
:param bool use_references: if False, forces all objects to be declared inline
instead of by referencing other components
:param kwargs: extra attributes for constructing the object;
if swagger_object_type is Parameter, ``name`` and ``in_`` should be provided
:return: the swagger object
:rtype: openapi.Parameter or openapi.Items or openapi.Schema or
openapi.SchemaRef
"""
return NotHandled
[docs]
def probe_field_inspectors(
self, field, swagger_object_type, use_references, **kwargs
):
"""Helper method for recursively probing `field_inspectors` to handle a given
field.
All arguments are the same as :meth:`.field_to_swagger_object`.
:rtype: openapi.Parameter or openapi.Items or openapi.Schema or
openapi.SchemaRef
"""
return self.probe_inspectors(
self.field_inspectors,
"field_to_swagger_object",
field,
{"field_inspectors": self.field_inspectors},
swagger_object_type=swagger_object_type,
use_references=use_references,
**kwargs,
)
[docs]
def _get_partial_types(self, field, swagger_object_type, use_references, **kwargs):
"""Helper method to extract generic information from a field and return a
partial constructor for the appropriate openapi object.
All arguments are the same as :meth:`.field_to_swagger_object`.
The return value is a tuple consisting of:
* a function for constructing objects of `swagger_object_type`; its prototype
is: ::
def SwaggerType(existing_object=None, **instance_kwargs):
This function creates an instance of `swagger_object_type`, passing the
following attributes to its init, in order of precedence:
- arguments specified by the ``kwargs`` parameter of
:meth:`._get_partial_types`
- ``instance_kwargs`` passed to the constructor function
- ``title``, ``description``, ``required``, ``x-nullable`` and ``default``
inferred from the field, where appropriate
If ``existing_object`` is not ``None``, it is updated instead of creating a
new object.
* a type that should be used for child objects if `field` is of an array type.
This can currently have two values:
- :class:`.Schema` if `swagger_object_type` is :class:`.Schema`
- :class:`.Items` if `swagger_object_type` is :class:`.Parameter` or
:class:`.Items`
:rtype: (function,type[openapi.Schema] or type[openapi.Items])
"""
assert swagger_object_type in (openapi.Schema, openapi.Parameter, openapi.Items)
assert not isinstance(field, openapi.SwaggerDict), (
"passed field is already a SwaggerDict object"
)
title = force_real_str(field.label) if field.label else None
title = (
title if swagger_object_type == openapi.Schema else None
) # only Schema has title
help_text = getattr(field, "help_text", None)
description = force_real_str(help_text) if help_text else None
description = (
description if swagger_object_type != openapi.Items else None
) # Items has no description either
def SwaggerType(existing_object=None, use_field_title=True, **instance_kwargs):
if (
"required" not in instance_kwargs
and swagger_object_type == openapi.Parameter
):
instance_kwargs["required"] = field.required
if (
"default" not in instance_kwargs
and swagger_object_type != openapi.Items
):
default = get_field_default(field)
if default not in (None, serializers.empty):
instance_kwargs["default"] = default
if (
use_field_title
and instance_kwargs.get("type", None) != openapi.TYPE_ARRAY
):
instance_kwargs.setdefault("title", title)
if description is not None:
instance_kwargs.setdefault("description", description)
if getattr(field, "allow_null", None):
instance_kwargs["x_nullable"] = True
instance_kwargs.update(kwargs)
if existing_object is not None:
assert isinstance(existing_object, swagger_object_type)
for key, val in sorted(instance_kwargs.items()):
setattr(existing_object, key, val)
result = existing_object
else:
result = swagger_object_type(**instance_kwargs)
# Provide an option to add manual parameters to a schema
# for example, to add examples
if swagger_object_type == openapi.Schema:
self.add_manual_fields(field, result)
return result
# arrays in Schema have Schema elements, arrays in Parameter and Items have
# Items elements
child_swagger_type = (
openapi.Schema if swagger_object_type == openapi.Schema else openapi.Items
)
return SwaggerType, child_swagger_type
class SerializerInspector(FieldInspector):
[docs]
def get_schema(self, serializer):
"""Convert a DRF Serializer instance to an :class:`.openapi.Schema`.
Should return :data:`.NotHandled` if this inspector does not know how to handle
the given `serializer`.
:param serializers.BaseSerializer serializer: the ``Serializer`` instance
:rtype: openapi.Schema
"""
return NotHandled
[docs]
def get_request_parameters(self, serializer, in_):
"""Convert a DRF serializer into a list of :class:`.Parameter`\\ s.
Should return :data:`.NotHandled` if this inspector does not know how to handle
the given `serializer`.
:param serializers.BaseSerializer serializer: the ``Serializer`` instance
:param str in_: the location of the parameters, one of the `openapi.IN_*`
constants
:rtype: list[openapi.Parameter]
"""
return NotHandled
class ViewInspector(BaseInspector):
body_methods = (
"PUT",
"PATCH",
"POST",
"DELETE",
) #: methods that are allowed to have a request body
#: methods that are assumed to require a request body determined by the view's
# ``serializer_class``
implicit_body_methods = ("PUT", "PATCH", "POST")
#: methods which are assumed to return a list of objects when present on non-detail
# endpoints
implicit_list_response_methods = ("GET",)
# real values set in __init__ to prevent import errors
field_inspectors = [] #:
filter_inspectors = [] #:
paginator_inspectors = [] #:
def __init__(self, view, path, method, components, request, overrides):
"""
Inspector class responsible for providing :class:`.Operation` definitions given
a view, path and method.
:param dict overrides: manual overrides as passed to
:func:`@swagger_auto_schema <.swagger_auto_schema>`
"""
super(ViewInspector, self).__init__(view, path, method, components, request)
self.overrides = overrides
self._prepend_inspector_overrides("field_inspectors")
self._prepend_inspector_overrides("filter_inspectors")
self._prepend_inspector_overrides("paginator_inspectors")
[docs]
def _prepend_inspector_overrides(self, inspectors):
extra_inspectors = self.overrides.get(inspectors, None)
if extra_inspectors:
default_inspectors = [
insp
for insp in getattr(self, inspectors)
if insp not in extra_inspectors
]
setattr(self, inspectors, extra_inspectors + default_inspectors)
[docs]
def get_operation(self, operation_keys):
"""Get an :class:`.Operation` for the given API endpoint (path, method).
This includes query, body parameters and response schemas.
:param tuple[str] operation_keys: an array of keys describing the hierarchical
layout of this view in the API; e.g. ``('snippets', 'list')``,
``('snippets', 'retrieve')``, etc.
:rtype: openapi.Operation
"""
raise NotImplementedError("ViewInspector must implement get_operation()!")
[docs]
def is_list_view(self):
"""Determine whether this view is a list or a detail view. The difference
between the two is that detail views depend on a pk/id path parameter. Note that
a non-detail view does not necessarily imply a list response
(:meth:`.has_list_response`), nor are list responses limited to non-detail
views.
For example, one might have a `/topic/<pk>/posts` endpoint which is a detail
view that has a list response.
:rtype: bool"""
return is_list_view(self.path, self.method, self.view)
[docs]
def has_list_response(self):
"""Determine whether this view returns multiple objects. By default this is any
non-detail view (see :meth:`.is_list_view`) whose request method is one of
:attr:`.implicit_list_response_methods`.
:rtype: bool
"""
return self.is_list_view() and (
self.method.upper() in self.implicit_list_response_methods
)
[docs]
def should_filter(self):
"""Determine whether filter backend parameters should be included for this
request.
:rtype: bool
"""
return getattr(self.view, "filter_backends", None) and self.has_list_response()
[docs]
def get_filter_parameters(self):
"""Return the parameters added to the view by its filter backends.
:rtype: list[openapi.Parameter]
"""
if not self.should_filter():
return []
fields = []
for filter_backend in getattr(self.view, "filter_backends"):
fields += (
self.probe_inspectors(
self.filter_inspectors, "get_filter_parameters", filter_backend()
)
or []
)
return fields
[docs]
def should_page(self):
"""Determine whether paging parameters and structure should be added to this
operation's request and response.
:rtype: bool
"""
return getattr(self.view, "paginator", None) and self.has_list_response()
[docs]
def serializer_to_schema(self, serializer):
"""Convert a serializer to an OpenAPI :class:`.Schema`.
:param serializers.BaseSerializer serializer: the ``Serializer`` instance
:returns: the converted :class:`.Schema`, or ``None`` in case of an unknown
serializer
:rtype: openapi.Schema or openapi.SchemaRef
"""
return self.probe_inspectors(
self.field_inspectors,
"get_schema",
serializer,
{"field_inspectors": self.field_inspectors},
)
[docs]
def serializer_to_parameters(self, serializer, in_):
"""Convert a serializer to a possibly empty list of :class:`.Parameter`\\ s.
:param serializers.BaseSerializer serializer: the ``Serializer`` instance
:param str in_: the location of the parameters, one of the `openapi.IN_*`
constants
:rtype: list[openapi.Parameter]
"""
return (
self.probe_inspectors(
self.field_inspectors,
"get_request_parameters",
serializer,
{"field_inspectors": self.field_inspectors},
in_=in_,
)
or []
)
[docs]
def get_paginated_response(self, response_schema):
"""Add appropriate paging fields to a response :class:`.Schema`.
:param openapi.Schema response_schema: the response schema that must be paged.
:returns: the paginated response class:`.Schema`, or ``None`` in case of an
unknown pagination scheme
:rtype: openapi.Schema
"""
return self.probe_inspectors(
self.paginator_inspectors,
"get_paginated_response",
getattr(self.view, "paginator"),
response_schema=response_schema,
)