Add OpenAPI 3.1 webhook support

drf_spectacular/generators.py

@@ -11,7 +11,7 @@
     add_trace_message, error, get_override, reset_generator_stats, warn,
 )
 from drf_spectacular.extensions import OpenApiViewExtension
-from drf_spectacular.openapi import AutoSchema
+from drf_spectacular.openapi import AutoSchema, process_webhooks
 from drf_spectacular.plumbing import (
     ComponentRegistry, alpha_operation_sorter, build_root_object, camelize_operation, get_class,
     is_versioning_supported, modify_for_versioning, normalize_result_object,
@@ -102,6 +102,7 @@ def __init__(self, *args, **kwargs):
         self.registry = ComponentRegistry()
         self.api_version = kwargs.pop('api_version', None)
         self.inspector = None
+        self.webhooks = kwargs.pop('webhooks', None)
         super().__init__(*args, **kwargs)
 
     def coerce_path(self, path, method, view):
@@ -183,6 +184,11 @@ def _initialise_endpoints(self):
             self.inspector = self.endpoint_inspector_cls(self.patterns, self.urlconf)
             self.endpoints = self.inspector.get_api_endpoints()
 
+    def _initialise_webhooks(self):
+        if self.webhooks is None:
+            if spectacular_settings.OAS_VERSION.startswith('3.1'):
+                self.webhooks = spectacular_settings.WEBHOOKS
+
     def _get_paths_and_endpoints(self):
         """
         Generate (path, method, view) given (path, method, callback) for paths.
@@ -274,12 +280,17 @@ def parse(self, input_request, public):
 
         return result
 
+    def get_webhooks(self):
+        self._initialise_webhooks()
+        return self.webhooks or []
+
     def get_schema(self, request=None, public=False):
         """ Generate a OpenAPI schema. """
         reset_generator_stats()
         result = build_root_object(
             paths=self.parse(request, public),
             components=self.registry.build(spectacular_settings.APPEND_COMPONENTS),
+            webhooks=process_webhooks(self.get_webhooks(), self.registry),
             version=self.api_version or getattr(request, 'version', None),
         )
         for hook in spectacular_settings.POSTPROCESSING_HOOKS:

drf_spectacular/openapi.py

@@ -44,7 +44,7 @@
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import (
     Direction, OpenApiCallback, OpenApiExample, OpenApiParameter, OpenApiRequest, OpenApiResponse,
-    _SchemaType, _SerializerType,
+    OpenApiWebhook, _SchemaType, _SerializerType,
 )
 
 
@@ -1619,3 +1619,62 @@ def resolve_serializer(
                 del self.registry[component]
                 return ResolvedComponent(None, None)  # sentinel
             return component
+
+
+def process_webhooks(webhooks: List[OpenApiWebhook], registry: ComponentRegistry):
+    """
+    Creates a mocked view for every webhook. The given extend_schema decorator then
+    specifies the expectations on the receiving end of the callback. Effectively
+    simulates a sub-schema from the opposing perspective via a virtual view definition.
+    """
+    result = {}
+
+    for webhook in webhooks:
+        if isinstance(webhook.decorator, dict):
+            methods = webhook.decorator
+        else:
+            methods = {'post': webhook.decorator}
+
+        path_items = {}
+
+        for method, decorator in methods.items():
+            # a dict indicates a raw schema; use directly
+            if isinstance(decorator, dict):
+                path_items[method.lower()] = decorator
+                continue
+
+            mocked_view = build_mocked_view(
+                method=method,
+                path="/",
+                extend_schema_decorator=decorator,
+                registry=registry,
+            )
+            operation = {}
+
+            description = mocked_view.schema.get_description()
+            if description:
+                operation['description'] = description
+
+            summary = mocked_view.schema.get_summary()
+            if summary:
+                operation['summary'] = summary
+
+            request_body = mocked_view.schema._get_request_body('response')
+            if request_body:
+                operation['requestBody'] = request_body
+
+            deprecated = mocked_view.schema.is_deprecated()
+            if deprecated:
+                operation['deprecated'] = deprecated
+
+            operation['responses'] = mocked_view.schema._get_response_bodies('request')
+
+            extensions = mocked_view.schema.get_extensions()
+            if extensions:
+                operation.update(sanitize_specification_extensions(extensions))
+
+            path_items[method.lower()] = operation
+
+        result[webhook.name] = path_items
+
+    return result

drf_spectacular/plumbing.py

@@ -477,7 +477,7 @@ def build_bearer_security_scheme_object(header_name, token_prefix, bearer_format
         }
 
 
-def build_root_object(paths, components, version) -> _SchemaType:
+def build_root_object(paths, components, webhooks, version) -> _SchemaType:
     settings = spectacular_settings
     if settings.VERSION and version:
         version = f'{settings.VERSION} ({version})'
@@ -508,6 +508,8 @@ def build_root_object(paths, components, version) -> _SchemaType:
         root['tags'] = settings.TAGS
     if settings.EXTERNAL_DOCS:
         root['externalDocs'] = settings.EXTERNAL_DOCS
+    if webhooks:
+        root['webhooks'] = webhooks
     return root
 
 

drf_spectacular/settings.py

@@ -199,6 +199,8 @@
     'SERVERS': [],
     # Tags defined in the global scope
     'TAGS': [],
+    # Optional: List of OpenAPI 3.1 webhooks.
+    'WEBHOOKS': None,
     # Optional: MUST contain 'url', may contain "description"
     'EXTERNAL_DOCS': {},
 

drf_spectacular/utils.py

@@ -316,6 +316,16 @@ def __init__(
         self.decorator = decorator
 
 
+class OpenApiWebhook(OpenApiSchemaBase):
+    def __init__(
+            self,
+            name: _StrOrPromise,
+            decorator: Union[Callable[[F], F], Dict[str, Callable[[F], F]], Dict[str, Any]],
+    ):
+        self.name = name
+        self.decorator = decorator
+
+
 def extend_schema(
         operation_id: Optional[str] = None,
         parameters: Optional[Sequence[Union[OpenApiParameter, _SerializerType]]] = None,

tests/__init__.py

@@ -61,7 +61,7 @@ def assert_equal(actual, expected):
     assert actual == expected and not diff, diff
 
 
-def generate_schema(route, viewset=None, view=None, view_function=None, patterns=None):
+def generate_schema(route, viewset=None, view=None, view_function=None, patterns=None, webhooks=None):
     from django.urls import path
     from rest_framework import routers
     from rest_framework.viewsets import ViewSetMixin
@@ -80,7 +80,7 @@ def generate_schema(route, viewset=None, view=None, view_function=None, patterns
     else:
         assert route is None and isinstance(patterns, list)
 
-    generator = SchemaGenerator(patterns=patterns)
+    generator = SchemaGenerator(patterns=patterns, webhooks=webhooks)
     schema = generator.get_schema(request=None, public=True)
     validate_schema(schema)  # make sure generated schemas are always valid
     return schema

tests/test_webhooks.py

@@ -0,0 +1,51 @@
+from unittest import mock
+
+import pytest
+from rest_framework import serializers
+
+from drf_spectacular.generators import SchemaGenerator
+from drf_spectacular.utils import OpenApiResponse, OpenApiWebhook, extend_schema
+from tests import assert_schema, generate_schema
+
+
+class EventSerializer(serializers.Serializer):
+    id = serializers.CharField(read_only=True)
+    change = serializers.CharField()
+    external_id = serializers.CharField(write_only=True)
+
+
+urlpatterns = []  # type: ignore
+openapi_webhooks = [
+    OpenApiWebhook(
+        name='SubscriptionEvent',
+        decorator=extend_schema(
+            summary="some summary",
+            description='pushes events to callbackUrl as "application/x-www-form-urlencoded"',
+            request={
+                'application/x-www-form-urlencoded': EventSerializer,
+            },
+            responses={
+                200: OpenApiResponse(description='event was successfully received'),
+                '4XX': OpenApiResponse(description='event will be retried shortly'),
+            },
+        ),
+    )
+]
+
+
+@mock.patch('drf_spectacular.settings.spectacular_settings.OAS_VERSION', '3.1.0')
+def test_webhooks(no_warnings):
+    assert_schema(
+        generate_schema(None, patterns=[], webhooks=openapi_webhooks),
+        'tests/test_webhooks.yml'
+    )
+
+
+@pytest.mark.urls(__name__)
+@mock.patch('drf_spectacular.settings.spectacular_settings.OAS_VERSION', '3.1.0')
+@mock.patch('drf_spectacular.settings.spectacular_settings.WEBHOOKS', openapi_webhooks)
+def test_webhooks_settings(no_warnings):
+    assert_schema(
+        SchemaGenerator().get_schema(request=None, public=True),
+        'tests/test_webhooks.yml'
+    )

tests/test_webhooks.yml

@@ -0,0 +1,38 @@
+openapi: 3.1.0
+info:
+  title: ''
+  version: 0.0.0
+paths: {}
+components:
+  schemas:
+    Event:
+      type: object
+      properties:
+        id:
+          type: string
+          readOnly: true
+        change:
+          type: string
+        external_id:
+          type: string
+          writeOnly: true
+      required:
+      - change
+      - external_id
+      - id
+webhooks:
+  SubscriptionEvent:
+    post:
+      description: pushes events to callbackUrl as "application/x-www-form-urlencoded"
+      summary: some summary
+      requestBody:
+        content:
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/Event'
+        required: true
+      responses:
+        '200':
+          description: event was successfully received
+        4XX:
+          description: event will be retried shortly