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 __future__ import annotations
import typing
from packaging.version import Version

import marshmallow
from marshmallow.utils import is_collection

from apispec import APISpec
from apispec.exceptions import APISpecError
from .field_converter import FieldConverterMixin
from .common import (

__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 Version|str 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: Version | str, schema_name_resolver, spec: APISpec, ) -> None: self.openapi_version = ( Version(openapi_version) if isinstance(openapi_version, str) else openapi_version ) self.schema_name_resolver = schema_name_resolver self.spec = spec self.init_attribute_functions() self.init_parameter_attribute_functions() # Schema references self.refs: dict = {} def init_parameter_attribute_functions(self) -> None: self.parameter_attribute_functions = [ self.field2required, self.list2param, ]
[docs] def add_parameter_attribute_function(self, func) -> None: """Method to add a field parameter function to the list of field parameter functions that will be called on a field to convert it to a field parameter. :param func func: the field parameter function to add The attribute function will be bound to the `OpenAPIConverter <apispec.ext.marshmallow.openapi.OpenAPIConverter>` instance. It will be called for each field in a schema with `self <apispec.ext.marshmallow.openapi.OpenAPIConverter>` and a `field <marshmallow.fields.Field>` instance positional arguments and `ret <dict>` keyword argument. May mutate `ret`. User added field parameter functions will be called after all built-in field parameter functions in the order they were added. """ bound_func = func.__get__(self) setattr(self, func.__name__, bound_func) self.parameter_attribute_functions.append(bound_func)
[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 dictionary will contain a JSON Schema Object representation of the schema. :param schema: schema to add to the spec """ try: schema_instance = resolve_schema_instance(schema) # If schema is a string and is not found in registry, # assume it is a schema reference except marshmallow.exceptions.RegistryError: return 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 as exc: 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) ) from exc 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, *, location, name: str = "body", required: bool = False, description: str | None = None, ): """Return an array of OpenAPI parameters given a given marshmallow :class:`Schema <marshmallow.Schema>`. If `location` 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 OpenAPI 3, only "query", "header", "path" or "cookie" are allowed for the location of parameters. "requestBody" is used when fields are in the body. """ location = __location_map__.get(location, location) # OAS 2 body parameter if location == "body": param = { "in": location, "required": required, "name": name, "schema": self.resolve_nested_schema(schema), } 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._field2parameter( field_obj, name=field_obj.data_key or field_name, location=location, ) for field_name, field_obj in fields.items() ]
def _field2parameter( self, field: marshmallow.fields.Field, *, name: str, location: str ) -> dict: """Return an OpenAPI parameter as a `dict`, given a marshmallow :class:`Field <marshmallow.Field>`. """ ret: dict = {"in": location, "name": name} prop = self.field2property(field) if self.openapi_version.major < 3: ret.update(prop) else: if "description" in prop: ret["description"] = prop.pop("description") if "deprecated" in prop: ret["deprecated"] = prop.pop("deprecated") ret["schema"] = prop for param_attr_func in self.parameter_attribute_functions: ret.update(param_attr_func(field, ret=ret)) return ret
[docs] def field2required( self, field: marshmallow.fields.Field, **kwargs: typing.Any ) -> dict: """Return the dictionary of OpenAPI parameter attributes for a required field. :param Field field: A marshmallow field. :rtype: dict """ ret = {} partial = getattr(field.parent, "partial", False) ret["required"] = field.required and ( not partial or (is_collection(partial) and not in partial) # type:ignore ) return ret
[docs] def list2param(self, field: marshmallow.fields.Field, **kwargs: typing.Any) -> dict: """Return a dictionary of parameter properties from :class:`List <marshmallow.fields.List` fields. :param Field field: A marshmallow field. :rtype: dict """ ret: dict = {} if isinstance(field, marshmallow.fields.List): if self.openapi_version.major < 3: ret["collectionFormat"] = "multi" else: ret["explode"] = True ret["style"] = "form" 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) jsonschema = self.fields2jsonschema(fields, partial=partial) if hasattr(Meta, "title"): jsonschema["title"] = Meta.title if hasattr(Meta, "description"): jsonschema["description"] = Meta.description if hasattr(Meta, "unknown") and Meta.unknown != marshmallow.EXCLUDE: jsonschema["additionalProperties"] = Meta.unknown == marshmallow.INCLUDE return jsonschema
[docs] def fields2jsonschema(self, fields, *, 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|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": {}} for field_name, field_obj in fields.items(): observed_field_name = field_obj.data_key or field_name prop = self.field2property(field_obj) jsonschema["properties"][observed_field_name] = prop 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 = self.spec.components.get_ref("schema", self.refs[schema_key]) if getattr(schema, "many", False): return {"type": "array", "items": ref_schema} return ref_schema