Source code for apispec.ext.marshmallow.openapi

"""Utilities for generating OpenAPI Specification (fka Swagger) entities from
marshmallow :class:`Schemas <marshmallow.Schema>` and :class:`Fields <marshmallow.fields.Field>`.

.. warning::

    This module is treated as private API.
    Users should not need to use this module directly.
from collections import OrderedDict

import marshmallow
from marshmallow.utils import is_collection

from apispec.utils import OpenAPIVersion, build_reference
from apispec.exceptions import APISpecError
from .field_converter import FieldConverterMixin
from .common import (

    [int(part) for part in marshmallow.__version__.split(".") if part.isdigit()]

__location_map__ = {
    "match_info": "path",
    "query": "query",
    "querystring": "query",
    "json": "body",
    "headers": "header",
    "cookies": "cookie",
    "form": "formData",
    "files": "formData",

[docs]class OpenAPIConverter(FieldConverterMixin): """Adds methods for generating OpenAPI specification from marshmallow schemas and fields. :param str|OpenAPIVersion openapi_version: The OpenAPI version to use. Should be in the form '2.x' or '3.x.x' to comply with the OpenAPI standard. :param callable schema_name_resolver: Callable to generate the schema definition name. Receives the `Schema` class and returns the name to be used in refs within the generated spec. When working with circular referencing this function must must not return `None` for schemas in a circular reference chain. :param APISpec spec: An initalied spec. Nested schemas will be added to the spec """ def __init__(self, openapi_version, schema_name_resolver, spec): self.openapi_version = OpenAPIVersion(openapi_version) self.schema_name_resolver = schema_name_resolver self.spec = spec self.init_attribute_functions() # Schema references self.refs = {} @staticmethod def _observed_name(field, name): """Adjust field name to reflect `dump_to` and `load_from` attributes. :param Field field: A marshmallow field. :param str name: Field name :rtype: str """ if MARSHMALLOW_VERSION_INFO[0] < 3: # use getattr in case we're running against older versions of marshmallow. dump_to = getattr(field, "dump_to", None) load_from = getattr(field, "load_from", None) return dump_to or load_from or name return field.data_key or name
[docs] def resolve_nested_schema(self, schema): """Return the OpenAPI representation of a marshmallow Schema. Adds the schema to the spec if it isn't already present. Typically will return a dictionary with the reference to the schema's path in the spec unless the `schema_name_resolver` returns `None`, in which case the returned dictoinary will contain a JSON Schema Object representation of the schema. :param schema: schema to add to the spec """ schema_instance = resolve_schema_instance(schema) schema_key = make_schema_key(schema_instance) if schema_key not in self.refs: name = self.schema_name_resolver(schema) if not name: try: json_schema = self.schema2jsonschema(schema_instance) except RuntimeError: raise APISpecError( "Name resolver returned None for schema {schema} which is " "part of a chain of circular referencing schemas. Please" " ensure that the schema_name_resolver passed to" " MarshmallowPlugin returns a string for all circular" " referencing schemas.".format(schema=schema) ) if getattr(schema, "many", False): return {"type": "array", "items": json_schema} return json_schema name = get_unique_schema_name(self.spec.components, name) self.spec.components.schema(name, schema=schema) return self.get_ref_dict(schema_instance)
[docs] def schema2parameters( self, schema, *, default_in="body", name="body", required=False, description=None ): """Return an array of OpenAPI parameters given a given marshmallow :class:`Schema <marshmallow.Schema>`. If `default_in` is "body", then return an array of a single parameter; else return an array of a parameter for each included field in the :class:`Schema <marshmallow.Schema>`. """ openapi_default_in = __location_map__.get(default_in, default_in) if self.openapi_version.major < 3 and openapi_default_in == "body": prop = self.resolve_nested_schema(schema) param = { "in": openapi_default_in, "required": required, "name": name, "schema": prop, } if description: param["description"] = description return [param] assert not getattr( schema, "many", False ), "Schemas with many=True are only supported for 'json' location (aka 'in: body')" fields = get_fields(schema, exclude_dump_only=True) return self.fields2parameters(fields, default_in=default_in)
[docs] def fields2parameters(self, fields, *, default_in): """Return an array of OpenAPI parameters given a mapping between field names and :class:`Field <marshmallow.Field>` objects. If `default_in` is "body", then return an array of a single parameter; else return an array of a parameter for each included field in the :class:`Schema <marshmallow.Schema>`. In OpenAPI3, only "query", "header", "path" or "cookie" are allowed for the location of parameters. In OpenAPI 3, "requestBody" is used when fields are in the body. This function always returns a list, with a parameter for each included field in the :class:`Schema <marshmallow.Schema>`. """ parameters = [] body_param = None for field_name, field_obj in fields.items(): if field_obj.dump_only: continue param = self.field2parameter( field_obj, name=self._observed_name(field_obj, field_name), default_in=default_in, ) if ( self.openapi_version.major < 3 and param["in"] == "body" and body_param is not None ): body_param["schema"]["properties"].update(param["schema"]["properties"]) required_fields = param["schema"].get("required", []) if required_fields: body_param["schema"].setdefault("required", []).extend( required_fields ) else: if self.openapi_version.major < 3 and param["in"] == "body": body_param = param parameters.append(param) return parameters
[docs] def field2parameter(self, field, *, name, default_in): """Return an OpenAPI parameter as a `dict`, given a marshmallow :class:`Field <marshmallow.Field>`. """ location = field.metadata.get("location", None) prop = self.field2property(field) return self.property2parameter( prop, name=name, required=field.required, multiple=isinstance(field, marshmallow.fields.List), location=location, default_in=default_in, )
[docs] def property2parameter( self, prop, *, name, required, multiple, location, default_in ): """Return the Parameter Object definition for a JSON Schema property. :param dict prop: JSON Schema property :param str name: Field name :param bool required: Parameter is required :param bool multiple: Parameter is repeated :param str location: Location to look for ``name`` :param str default_in: Default location to look for ``name`` :raise: TranslationError if arg object cannot be translated to a Parameter Object schema. :rtype: dict, a Parameter Object """ openapi_default_in = __location_map__.get(default_in, default_in) openapi_location = __location_map__.get(location, openapi_default_in) ret = {"in": openapi_location, "name": name} if openapi_location == "body": ret["required"] = False ret["name"] = "body" ret["schema"] = {"type": "object", "properties": {name: prop}} if required: ret["schema"]["required"] = [name] else: ret["required"] = required if self.openapi_version.major < 3: if multiple: ret["collectionFormat"] = "multi" ret.update(prop) else: if multiple: ret["explode"] = True ret["style"] = "form" if prop.get("description", None): ret["description"] = prop.pop("description") ret["schema"] = prop return ret
[docs] def schema2jsonschema(self, schema): """Return the JSON Schema Object for a given marshmallow :class:`Schema <marshmallow.Schema>` instance. Schema may optionally provide the ``title`` and ``description`` class Meta options. :param Schema schema: A marshmallow Schema instance :rtype: dict, a JSON Schema Object """ fields = get_fields(schema) Meta = getattr(schema, "Meta", None) partial = getattr(schema, "partial", None) ordered = getattr(schema, "ordered", False) jsonschema = self.fields2jsonschema(fields, partial=partial, ordered=ordered) if hasattr(Meta, "title"): jsonschema["title"] = Meta.title if hasattr(Meta, "description"): jsonschema["description"] = Meta.description return jsonschema
[docs] def fields2jsonschema(self, fields, *, ordered=False, partial=None): """Return the JSON Schema Object given a mapping between field names and :class:`Field <marshmallow.Field>` objects. :param dict fields: A dictionary of field name field object pairs :param bool ordered: Whether to preserve the order in which fields were declared :param bool|tuple partial: Whether to override a field's required flag. If `True` no fields will be set as required. If an iterable fields in the iterable will not be marked as required. :rtype: dict, a JSON Schema Object """ jsonschema = {"type": "object", "properties": OrderedDict() if ordered else {}} for field_name, field_obj in fields.items(): observed_field_name = self._observed_name(field_obj, field_name) property = self.field2property(field_obj) jsonschema["properties"][observed_field_name] = property if field_obj.required: if not partial or ( is_collection(partial) and field_name not in partial ): jsonschema.setdefault("required", []).append(observed_field_name) if "required" in jsonschema: jsonschema["required"].sort() return jsonschema
[docs] def get_ref_dict(self, schema): """Method to create a dictionary containing a JSON reference to the schema in the spec """ schema_key = make_schema_key(schema) ref_schema = build_reference( "schema", self.openapi_version.major, self.refs[schema_key] ) if getattr(schema, "many", False): return {"type": "array", "items": ref_schema} return ref_schema