From 2c08679e33043b05a15444554ac5f8f95b7c6216 Mon Sep 17 00:00:00 2001 From: Matthias Dellweg Date: Fri, 7 Jun 2024 13:27:20 +0200 Subject: [PATCH] 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 | 4 ++ pulpcore/openapi/__init__.py | 111 +++++------------------------------ pulpcore/openapi/hooks.py | 33 +++++++++++ 3 files changed, 52 insertions(+), 96 deletions(-) create mode 100644 pulpcore/openapi/hooks.py diff --git a/pulpcore/app/settings.py b/pulpcore/app/settings.py index 6ba98056e4..7024a58d6c 100644 --- a/pulpcore/app/settings.py +++ b/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", diff --git a/pulpcore/openapi/__init__.py b/pulpcore/openapi/__init__.py index 4836952d5b..5b960af11a 100644 --- a/pulpcore/openapi/__init__.py +++ b/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): diff --git a/pulpcore/openapi/hooks.py b/pulpcore/openapi/hooks.py new file mode 100644 index 0000000000..3c03ac6fec --- /dev/null +++ b/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