Fixes #20638: Document bulk create support in OpenAPI schema

POST operations on NetBoxModelViewSet endpoints accept both single
objects and arrays, but the schema only documented single objects.
This prevented API client generators from producing correct code.

Add explicit bulk_create_enabled flag to NetBoxModelViewSet and
update schema generation to emit oneOf for these endpoints.
This commit is contained in:
Jason Novinger
2025-11-10 05:52:52 -06:00
parent 9723a2f0ad
commit 773d86dd85
4 changed files with 2924 additions and 268 deletions

File diff suppressed because it is too large Load Diff

View File

@@ -49,6 +49,18 @@ class ChoiceFieldFix(OpenApiSerializerFieldExtension):
)
def viewset_handles_bulk_create(view):
"""
Check if viewset explicitly declares bulk create support.
Viewsets opt-in to bulk create by setting bulk_create_enabled = True.
This allows POST operations to accept either single objects or arrays.
Refs: #20638
"""
return getattr(view, 'bulk_create_enabled', False)
class NetBoxAutoSchema(AutoSchema):
"""
Overrides to drf_spectacular.openapi.AutoSchema to fix following issues:
@@ -106,17 +118,36 @@ class NetBoxAutoSchema(AutoSchema):
if self.is_bulk_action:
return type(serializer)(many=True)
# For create operations (POST to list endpoints) on viewsets that support
# bulk create, mark the serializer so we can generate oneOf schema.
# The 'create' action is only used for POST to collection endpoints.
# Refs: #20638
if (
getattr(self.view, 'action', None) == 'create' and
viewset_handles_bulk_create(self.view)
):
# Mark this serializer as supporting array input
# This will be used in _get_request_for_media_type()
if serializer is not None:
serializer._netbox_supports_array = True
# handle mapping for Writable serializers - adapted from dansheps original code
# for drf-yasg
if serializer is not None and self.method in WRITABLE_ACTIONS:
writable_class = self.get_writable_class(serializer)
if writable_class is not None:
# Preserve the array support marker when creating writable serializer
supports_array = getattr(serializer, '_netbox_supports_array', False)
if hasattr(serializer, "child"):
child_serializer = self.get_writable_class(serializer.child)
serializer = writable_class(context=serializer.context, child=child_serializer)
else:
serializer = writable_class(context=serializer.context)
if supports_array:
serializer._netbox_supports_array = True
return serializer
def get_response_serializers(self) -> typing.Any:
@@ -128,6 +159,35 @@ class NetBoxAutoSchema(AutoSchema):
return response_serializers
def _get_request_for_media_type(self, serializer, direction):
"""
Override to generate oneOf schema for serializers that support both
single object and array input (NetBoxModelViewSet POST operations).
Refs: #20638
"""
# Get the standard schema first
schema, required = super()._get_request_for_media_type(serializer, direction)
# If this serializer supports arrays (marked in get_request_serializer),
# wrap the schema in oneOf to allow single object OR array
if (
direction == 'request' and
schema is not None and
getattr(serializer, '_netbox_supports_array', False)
):
return {
'oneOf': [
schema, # Single object
{
'type': 'array',
'items': schema, # Array of objects
}
]
}, required
return schema, required
def _get_serializer_name(self, serializer, direction, bypass_extensions=False) -> str:
name = super()._get_serializer_name(serializer, direction, bypass_extensions)

View File

@@ -0,0 +1,108 @@
"""
Unit tests for OpenAPI schema generation.
Refs: #20638
"""
import json
from django.test import TestCase
class OpenAPISchemaTestCase(TestCase):
"""Tests for OpenAPI schema generation."""
def setUp(self):
"""Fetch schema via API endpoint."""
response = self.client.get('/api/schema/', {'format': 'json'})
self.assertEqual(response.status_code, 200)
self.schema = json.loads(response.content)
def test_post_operation_documents_single_or_array(self):
"""
POST operations on NetBoxModelViewSet endpoints should document
support for both single objects and arrays via oneOf.
Refs: #20638
"""
# Test representative endpoints across different apps
test_paths = [
'/api/core/data-sources/',
'/api/dcim/sites/',
'/api/users/users/',
'/api/ipam/ip-addresses/',
]
for path in test_paths:
with self.subTest(path=path):
operation = self.schema['paths'][path]['post']
# Get the request body schema
request_schema = operation['requestBody']['content']['application/json']['schema']
# Should have oneOf with two options
self.assertIn('oneOf', request_schema, f"POST {path} should have oneOf schema")
self.assertEqual(
len(request_schema['oneOf']), 2,
f"POST {path} oneOf should have exactly 2 options"
)
# First option: single object (has $ref or properties)
single_schema = request_schema['oneOf'][0]
self.assertTrue(
'$ref' in single_schema or 'properties' in single_schema,
f"POST {path} first oneOf option should be single object"
)
# Second option: array of objects
array_schema = request_schema['oneOf'][1]
self.assertEqual(
array_schema['type'], 'array',
f"POST {path} second oneOf option should be array"
)
self.assertIn('items', array_schema, f"POST {path} array should have items")
def test_bulk_update_operations_require_array_only(self):
"""
Bulk update/patch operations should require arrays only, not oneOf.
They don't support single object input.
Refs: #20638
"""
test_paths = [
'/api/dcim/sites/',
'/api/users/users/',
]
for path in test_paths:
for method in ['put', 'patch']:
with self.subTest(path=path, method=method):
operation = self.schema['paths'][path][method]
request_schema = operation['requestBody']['content']['application/json']['schema']
# Should be array-only, not oneOf
self.assertNotIn(
'oneOf', request_schema,
f"{method.upper()} {path} should NOT have oneOf (array-only)"
)
self.assertEqual(
request_schema['type'], 'array',
f"{method.upper()} {path} should require array"
)
self.assertIn(
'items', request_schema,
f"{method.upper()} {path} array should have items"
)
def test_bulk_delete_requires_array(self):
"""
Bulk delete operations should require arrays.
Refs: #20638
"""
path = '/api/dcim/sites/'
operation = self.schema['paths'][path]['delete']
request_schema = operation['requestBody']['content']['application/json']['schema']
# Should be array-only
self.assertNotIn('oneOf', request_schema, "DELETE should NOT have oneOf")
self.assertEqual(request_schema['type'], 'array', "DELETE should require array")
self.assertIn('items', request_schema, "DELETE array should have items")

View File

@@ -128,6 +128,10 @@ class NetBoxModelViewSet(
qs = super().get_queryset()
return reapply_model_ordering(qs)
# Declare that this viewset accepts either single objects or arrays for creation
# Refs: #20638
bulk_create_enabled = True
def get_serializer(self, *args, **kwargs):
# If a list of objects has been provided, initialize the serializer with many=True
if isinstance(kwargs.get('data', {}), list):