Refactor openapi generation

Use postprocessing hooks instead of monkeypatching inner plumbings of
drf-spectacular. This should vastly improve stability and the ability to
upgrade spectacular.

[noissue]

pulpcore/app/settings.py

@@ -290,6 +290,10 @@
     "COMPONENT_NO_READ_ONLY_REQUIRED": True,
     "GENERIC_ADDITIONAL_PROPERTIES": None,
     "DISABLE_ERRORS_AND_WARNINGS": not DEBUG,
+    "POSTPROCESSING_HOOKS": [
+        "drf_spectacular.hooks.postprocess_schema_enums",
+        "pulpcore.openapi.hooks.add_info_hook",
+    ],
     "TITLE": "Pulp 3 API",
     "DESCRIPTION": "Fetch, Upload, Organize, and Distribute Software Packages",
     "VERSION": "v3",

pulpcore/openapi/__init__.py

@@ -5,6 +5,7 @@
 
 from django.conf import settings
 from django.core.exceptions import FieldDoesNotExist
+from django.http import HttpRequest
 from django.utils.html import strip_tags
 from drf_spectacular.drainage import reset_generator_stats
 from drf_spectacular.generators import SchemaGenerator
@@ -26,6 +27,7 @@
 from drf_spectacular.extensions import OpenApiAuthenticationExtension
 from rest_framework import mixins, serializers
 from rest_framework.exceptions import ParseError
+from rest_framework.request import Request
 from rest_framework.schemas.utils import get_pk_description
 
 from pulpcore.app.apps import pulp_plugin_configs
@@ -215,78 +217,6 @@ def _get_request_body(self):
             request_body["required"] = True
         return request_body
 
-    def _resolve_path_parameters(self, variables):
-        """
-        Resolve path parameters.
-
-        Extended to omit undesired warns.
-        """
-        model = getattr(getattr(self.view, "queryset", None), "model", None)
-        parameters = []
-
-        for variable in variables:
-            schema = build_basic_type(OpenApiTypes.STR)
-            description = ""
-
-            resolved_parameter = resolve_django_path_parameter(
-                self.path_regex,
-                variable,
-                self.map_renderers("format"),
-            )
-            if not resolved_parameter:
-                resolved_parameter = resolve_regex_path_parameter(self.path_regex, variable)
-
-            if resolved_parameter:
-                schema = resolved_parameter["schema"]
-            elif model:
-                try:
-                    model_field = model._meta.get_field(variable)
-                    schema = self._map_model_field(model_field, direction=None)
-                    # strip irrelevant meta data
-                    irrelevant_field_meta = ["readOnly", "writeOnly", "nullable", "default"]
-                    schema = {k: v for k, v in schema.items() if k not in irrelevant_field_meta}
-                    if "description" not in schema and model_field.primary_key:
-                        description = get_pk_description(model, model_field)
-                except FieldDoesNotExist:
-                    pass
-
-            # Used by the bindings to identify which param is the domain path
-            extensions = {}
-            if settings.DOMAIN_ENABLED and variable == "pulp_domain":
-                extensions["x-isDomain"] = True
-
-            parameters.append(
-                build_parameter_type(
-                    name=variable,
-                    location=OpenApiParameter.PATH,
-                    description=description,
-                    schema=schema,
-                    extensions=extensions,
-                )
-            )
-
-        return parameters
-
-    def resolve_serializer(self, serializer, direction):
-        """Serializer to component."""
-        component_schema = self._map_serializer(serializer, direction)
-        if not component_schema.get("properties", {}):
-            component = ResolvedComponent(
-                name=self._get_serializer_name(serializer, direction),
-                type=ResolvedComponent.SCHEMA,
-                object=serializer,
-            )
-            if component in self.registry:
-                return self.registry[component]
-
-            component.schema = component_schema
-            self.registry.register(component)
-
-        else:
-            component = super().resolve_serializer(serializer, direction)
-
-        return component
-
     def _get_response_bodies(self):
         """
         Handle response status code.
@@ -472,7 +402,10 @@ def parse(self, input_request, public):
 
     def get_schema(self, request=None, public=False):
         """Generate a OpenAPI schema."""
-        reset_generator_stats()
+        if request is None:
+            request = Request(HttpRequest())
+            request.META["SERVER_NAME"] = "localhost"
+            request.META["SERVER_PORT"] = "24817"
 
         apps = list(pulp_plugin_configs())
         if request and "plugin" in request.query_params:
@@ -484,32 +417,18 @@ def get_schema(self, request=None, public=False):
             if len(apps) != len(app_labels):
                 raise ParseError("Invalid component specified.")
             request.plugins = [app.name.split(".")[0] for app in apps]
+        request.apps = apps
 
-        result = build_root_object(
-            paths=self.parse(request, public),
-            components=self.registry.build(spectacular_settings.APPEND_COMPONENTS),
-            webhooks=process_webhooks(spectacular_settings.WEBHOOKS, self.registry),
-            version=self.api_version or getattr(request, "version", None),
-        )
-        for hook in spectacular_settings.POSTPROCESSING_HOOKS:
-            result = hook(result=result, generator=self, request=request, public=public)
-
-        # Basically I'm doing it to get pulp logo at redoc page
-        result["info"]["x-logo"] = {
-            "url": "https://pulp.plan.io/attachments/download/517478/pulp_logo_word_rectangle.svg"
-        }
-
-        # Adding plugin version config
-        result["info"]["x-pulp-app-versions"] = {app.label: app.version for app in apps}
+        request.bindings = "bindings" in request.query_params
 
-        # Add domain-settings value
-        result["info"]["x-pulp-domain-enabled"] = settings.DOMAIN_ENABLED
+        result = super().get_schema(request, public)
 
-        # Adding current host as server (it will provide a default value for the bindings)
-        server_url = "http://localhost:24817" if not request else request.build_absolute_uri("/")
-        result["servers"] = [{"url": server_url}]
-
-        return normalize_result_object(result)
+        if request.bindings:
+            # Undo the spectacular sanitization of operation ids
+            for path, path_spec in result["paths"].items():
+                for operation, operation_spec in path_spec.items():
+                    operation_spec["operationId"] = operation_spec.pop("x-copy-operationId")
+        return result
 
 
 class JSONHeaderRemoteAuthenticationScheme(OpenApiAuthenticationExtension):

pulpcore/openapi/hooks.py

@@ -0,0 +1,33 @@
+from django.conf import settings
+
+
+def add_info_hook(result, generator, request, **kwargs):
+    # Basically I'm doing it to get pulp logo at redoc page
+    result["info"]["x-logo"] = {
+        "url": "https://pulp.plan.io/attachments/download/517478/pulp_logo_word_rectangle.svg"
+    }
+
+    # Adding plugin version config
+    result["info"]["x-pulp-app-versions"] = {app.label: app.version for app in request.apps}
+
+    # Add domain-settings value
+    result["info"]["x-pulp-domain-enabled"] = settings.DOMAIN_ENABLED
+
+    # Add x-isDomain flag to domain path parameters
+    # Save a copy of the bad operationId in case we generate bindings
+    for path, path_spec in result["paths"].items():
+        for operation, operation_spec in path_spec.items():
+            if request.bindings:
+                # Keep the operation id before sanitization
+                operation_spec["x-copy-operationId"] = operation_spec["operationId"]
+            if settings.DOMAIN_ENABLED:
+                parameters = operation_spec.get("parameters")
+                if parameters:
+                    for parameter in parameters:
+                        if parameter["name"] == "pulp_domain" and parameter["in"] == "path":
+                            parameter["x-isDomain"] = True
+
+    # Adding current host as server (it will provide a default value for the bindings)
+    result["servers"] = [{"url": request.build_absolute_uri("/")}]
+
+    return result