Examine the possible ways to integrate with `drf-spectacular`

[surenkov]

Along with DRF's schema generators, it would be handy to have an adapter for def-spectacular, as it's becoming de-facto standard tool for OpenAPI schema generation in DRF.

[johnthagen]

drf-spectacular's maintainer has been open to upstreaming support for popular third party libraries: https://drf-spectacular.readthedocs.io/en/latest/readme.html

Likely the best way to support this, would be to upstream support into drf-spectacular itself.

[surenkov]

@johnthagen thank you for mentioning this.

I'll try to allocate a bit more time on this integration, but can't give any estimates... :(

[andreasnuesslein]

Ahoy there, so I came up with this:

from django_pydantic_field.v2.fields import PydanticSchemaField
from django_pydantic_field.v2.rest_framework import SchemaField
from rest_framework import serializers

class SpectacularSchemaField(SchemaField):
    def __init__(self, exclude_unset=True, *args, **kwargs):
        kwargs.pop("encoder", None)
        kwargs.pop("decoder", None)
        super().__init__(
            schema=self._spectacular_annotation["field"],
            exclude_unset=exclude_unset,
            *args,
            **kwargs,
        )

class MyModelSerializer(serializers.ModelSerializer):
    def build_standard_field(self, field_name, model_field):
        standard_field = super().build_standard_field(field_name, model_field)
        if isinstance(model_field, PydanticSchemaField):
            standard_field = (
                type(
                    model_field.schema.__name__ + "Serializer",
                    (SpectacularSchemaField,),
                    {"_spectacular_annotation": {"field": model_field.schema}},
                ),
            ) + standard_field[1:]
        return standard_field

    class Meta:
        abstract = True

and then just use MyModelSerializer and it SHOULD output the right serializer for the DRF and give you the right spectacular annotation. "works on my machine" :tm:

[andreasnuesslein]

PS this would be the semi-automatic version:

class _NoToGreatlyOtherSerializer(SpectacularSchemaField):
    _spectacular_annotation = {"field": app_schema.NoToGreatlyOtherSchema}

class ApproachVersionSerializer(serializers.ModelSerializer):
    impacts_other = _NoToGreatlyOtherSerializer()

[holtgrewe]

Something that helped me in my project, see below.

from drf_spectacular.drainage import set_override, warn
from drf_spectacular.extensions import OpenApiSerializerExtension
from drf_spectacular.plumbing import ResolvedComponent, build_basic_type
from drf_spectacular.types import OpenApiTypes
from pydantic.json_schema import model_json_schema

class DjangoPydanticFieldFix(OpenApiSerializerExtension):

    target_class = "django_pydantic_field.v2.rest_framework.fields.SchemaField"
    match_subclasses = True

    def get_name(self, auto_schema, direction):
        # due to the fact that it is complicated to pull out every field member BaseModel class
        # of the entry model, we simply use the class name as string for object. This hack may
        # create false positive warnings, so turn it off. However, this may suppress correct
        # warnings involving the entry class.
        set_override(self.target, "suppress_collision_warning", True)
        if typing.get_origin(self.target.schema) is list:
            inner_type = typing.get_args(self.target.schema)[0]
            return f"{inner_type.__name__}List"
        else:
            return super().get_name(auto_schema, direction)

    def map_serializer(self, auto_schema, direction):
        if typing.get_origin(self.target.schema) is list:
            inner_type = typing.get_args(self.target.schema)[0]
            if inner_type is str:
                schema = {
                    "type": "array",
                    "items": {
                        "type": "string",
                    },
                }
            elif issubclass(inner_type, Enum):
                inner_schema = {
                    "type": "string",
                    "title": inner_type.__name__,
                    "enum": [e.value for e in inner_type],
                }
                inner_schema_defs = inner_schema.pop("$defs", {})
                schema = {
                    "type": "array",
                    "title": f"{inner_schema['title']}List",
                    "items": inner_schema,
                }
                schema.update({"$defs": inner_schema_defs})
            else:
                inner_schema = model_json_schema(
                    inner_type, ref_template="#/components/schemas/{model}"
                )
                inner_schema_defs = inner_schema.pop("$defs", {})
                schema = {
                    "type": "array",
                    "title": f"{inner_schema['title']}List",
                    "items": inner_schema,
                }
                schema.update({"$defs": inner_schema_defs})
        elif issubclass(self.target.schema, Enum):
            return {
                "type": "string",
                "title": self.target.schema.__name__,
                "enum": [e.value for e in self.target.schema],
            }
        else:
            schema = model_json_schema(
                self.target.schema, ref_template="#/components/schemas/{model}"
            )

        # pull out potential sub-schemas and put them into component section
        for sub_name, sub_schema in schema.pop("$defs", {}).items():
            component = ResolvedComponent(
                name=sub_name,
                type=ResolvedComponent.SCHEMA,
                object=sub_name,
                schema=sub_schema,
            )
            auto_schema.registry.register_on_missing(component)

        return schema

[holtgrewe]

OK... my previous post did not work. One has to subclass the OpenApiSerializer**Field**Extension. Here is something that works for a couple of combinations of Optional, Union and list. However, it's not complete yet, only what I needed for my project.

from enum import Enum
from inspect import isclass
import typing

from drf_spectacular.drainage import set_override
from drf_spectacular.extensions import OpenApiSerializerFieldExtension
from drf_spectacular.plumbing import ResolvedComponent
import pydantic
from pydantic.json_schema import model_json_schema

def pydantic_to_json_schema(schema_arg: typing.Any) -> typing.Dict[str, typing.Any]:
    """Convert a Python/pydantic schema to a JSON schema."""
    if type(schema_arg) is type(int) or type(schema_arg) is type(float):
        return {
            "type": "number",
        }
    elif type(schema_arg) is type(str):
        return {
            "type": "string",
        }
    elif type(schema_arg) is type(None):
        return {
            "type": "null",
        }
    elif isclass(schema_arg) and issubclass(schema_arg, Enum):
        return {
            "type": "string",
            "title": schema_arg.__name__,
            "enum": [e.value for e in schema_arg],
        }
    elif (  # is typing.Optional[X]
        typing.get_origin(schema_arg) is typing.Union
    ):
        schema_arg = typing.get_args(schema_arg)[0]
        one_ofs = [
            pydantic_to_json_schema(arg_inner)
            for arg_inner in typing.get_args(schema_arg)
        ]
        defs = {}
        for one_of in one_ofs:
            defs.update(one_of.pop("$defs", {}))
        result = {
            "oneOf": one_ofs,
            "$defs": defs
        }
        return result
    elif typing.get_origin(schema_arg) is list:
        inner_schema = pydantic_to_json_schema(typing.get_args(schema_arg)[0])
        defs = inner_schema.pop("$defs", {})
        return {
            "type": "array",
            "items": inner_schema,
            "$defs": defs,
        }
    elif issubclass(schema_arg, Enum):
        return {
            "type": "string",
            "title": schema_arg.__name__,
            "enum": [e.value for e in schema_arg],
        }
    elif issubclass(schema_arg, pydantic.BaseModel):
        return model_json_schema(schema_arg, ref_template="#/components/schemas/{model}")
    else:
        raise ValueError(f"Unsupported schema type: {schema_arg}")

class DjangoPydanticFieldFix(OpenApiSerializerFieldExtension):

    target_class = "django_pydantic_field.v2.rest_framework.fields.SchemaField"
    match_subclasses = True

    def get_name(self):
        # due to the fact that it is complicated to pull out every field member BaseModel class
        # of the entry model, we simply use the class name as string for object. This hack may
        # create false positive warnings, so turn it off. However, this may suppress correct
        # warnings involving the entry class.
        set_override(self.target, "suppress_collision_warning", True)
        if typing.get_origin(self.target.schema) is list:
            inner_type = typing.get_args(self.target.schema)[0]
            return f"{inner_type.__name__}List"
        else:
            return super().get_name()

    def map_serializer_field(self, auto_schema, direction):
        _ = direction
        schema = pydantic_to_json_schema(self.target.schema)
        # pull out potential sub-schemas and put them into component section
        for sub_name, sub_schema in schema.pop("$defs", {}).items():
            component = ResolvedComponent(
                name=sub_name,
                type=ResolvedComponent.SCHEMA,
                object=sub_name,
                schema=sub_schema,
            )
            auto_schema.registry.register_on_missing(component)

        return schema

[leowonglaw]

This is my two cents

I needed just a pydantic field in my serializer, but on swagger it was showing as "string". Also using django_pydantic_field.rest_framework.AutoSchema raised an error:

  File "/usr/local/lib/python3.8/site-packages/django_pydantic_field/v2/rest_framework/openapi.py", line 27, in __init__
    super().__init__(tags, operation_id_base, component_name)
TypeError: __init__() takes 1 positional argument but 4 were given

My solution for having openAPI schema for a pydantic field in a serializer:

from drf_spectacular.utils import extend_schema_field

def custom_pydantic_field(pydantic_model):
    @extend_schema_field(pydantic_model)
    class MyPydanticField(SchemaField):
        pass
    return MyPydanticField(schema=pydantic_model)

class MyPydanticModel(BaseModel):
    is_enabled: bool = Field(default=True)

class MySerializer(serializers.ModelSerializer):
    review_filter = custom_pydantic_field(MyPydanticModel)

# no need for AutoSchema on views