allow plugins to override get_model_serializer

This commit is contained in:
Arthur 2026-05-07 08:46:48 -07:00 committed by Jeremy Stretch
parent 659d6d1f85
commit 0499bb7616
5 changed files with 121 additions and 0 deletions

View File

@ -32,6 +32,7 @@ DEFAULT_RESOURCE_PATHS = {
'graphql_schema': 'graphql.schema',
'menu': 'navigation.menu',
'menu_items': 'navigation.menu_items',
'serializer_resolver': 'api.serializers.serializer_resolver',
'template_extensions': 'template_content.template_extensions',
'user_preferences': 'preferences.preferences',
}
@ -80,6 +81,7 @@ class PluginConfig(AppConfig):
graphql_schema = None
menu = None
menu_items = None
serializer_resolver = None
template_extensions = None
user_preferences = None
events_pipeline = []
@ -134,6 +136,11 @@ class PluginConfig(AppConfig):
if user_preferences := self._load_resource('user_preferences'):
register_user_preferences(plugin_name, user_preferences)
# Register a serializer resolver (if defined) so utilities.api.get_serializer_for_model
# can resolve serializers for dynamically generated or otherwise non-conventional models.
if serializer_resolver := self._load_resource('serializer_resolver'):
register_serializer_resolver(serializer_resolver)
@classmethod
def validate(cls, user_config, netbox_version):

View File

@ -11,6 +11,7 @@ __all__ = (
'register_graphql_schema',
'register_menu',
'register_menu_items',
'register_serializer_resolver',
'register_template_extensions',
'register_user_preferences',
)
@ -82,3 +83,22 @@ def register_user_preferences(plugin_name, preferences):
Register a list of user preferences defined by a plugin.
"""
registry['plugins']['preferences'][plugin_name] = preferences
def register_serializer_resolver(resolver):
"""
Register a callable that returns a DRF serializer class for a model, or
None if the resolver does not handle the model. Resolvers are tried in
registration order before the default import-path lookup performed by
utilities.api.get_serializer_for_model().
This is the supported extension point for plugins whose models are
generated dynamically (and therefore have no importable serializer at
the {app_label}.api.serializers.{Model}Serializer path) or that need
to override serializer resolution for specific models.
Resolver signature: resolver(model, prefix='') -> serializer class or None
"""
if not callable(resolver):
raise TypeError(_("Serializer resolver must be callable"))
registry['serializer_resolvers'].append(resolver)

View File

@ -44,6 +44,7 @@ registry = Registry({
'plugins': dict(),
'request_processors': list(),
'search': dict(),
'serializer_resolvers': list(),
'system_jobs': dict(),
'tables': collections.defaultdict(dict),
'views': collections.defaultdict(dict),

View File

@ -17,6 +17,7 @@ from rest_framework.views import get_view_name as drf_get_view_name
from extras.constants import HTTP_CONTENT_TYPE_JSON
from netbox.api.exceptions import GraphQLTypeNotFound, SerializerNotFound
from netbox.api.fields import RelatedObjectCountField
from netbox.registry import registry
from .query import count_related, dict_to_filter_params
from .string import title
@ -45,7 +46,18 @@ class IsSuperuser(BasePermission):
def get_serializer_for_model(model, prefix=''):
"""
Return the appropriate REST API serializer for the given model.
Plugins may register custom resolvers via
netbox.plugins.register_serializer_resolver() to handle dynamically
generated models or to override serializer resolution for specific
models. Resolvers run in registration order; the first non-None return
wins. If no resolver matches, the default import-path lookup runs.
"""
for resolver in registry['serializer_resolvers']:
serializer = resolver(model, prefix=prefix)
if serializer is not None:
return serializer
app_label, model_name = model._meta.label.split('.')
serializer_name = f'{app_label}.api.serializers.{prefix}{model_name}Serializer'
try:

View File

@ -476,6 +476,87 @@ class GetPrefetchesForSerializerTestCase(TestCase):
)
class SerializerResolverRegistryTestCase(TestCase):
"""
Verify that registered serializer resolvers are consulted before the
default import-path lookup in get_serializer_for_model().
"""
def setUp(self):
from netbox.registry import registry
# Snapshot and clear the resolver list so each test starts from a
# known state and can't leak resolvers into the rest of the suite.
self._saved_resolvers = list(registry['serializer_resolvers'])
registry['serializer_resolvers'].clear()
def tearDown(self):
from netbox.registry import registry
registry['serializer_resolvers'].clear()
registry['serializer_resolvers'].extend(self._saved_resolvers)
def test_default_lookup_when_no_resolvers_registered(self):
from dcim.api.serializers import SiteSerializer
from utilities.api import get_serializer_for_model
self.assertIs(get_serializer_for_model(Site), SiteSerializer)
def test_registered_resolver_overrides_default(self):
from netbox.plugins import register_serializer_resolver
from utilities.api import get_serializer_for_model
sentinel = object()
register_serializer_resolver(lambda model, prefix='': sentinel)
self.assertIs(get_serializer_for_model(Site), sentinel)
def test_resolver_returning_none_falls_through_to_default(self):
from dcim.api.serializers import SiteSerializer
from netbox.plugins import register_serializer_resolver
from utilities.api import get_serializer_for_model
register_serializer_resolver(lambda model, prefix='': None)
self.assertIs(get_serializer_for_model(Site), SiteSerializer)
def test_resolvers_tried_in_registration_order(self):
from netbox.plugins import register_serializer_resolver
from utilities.api import get_serializer_for_model
first = object()
second = object()
# First resolver only handles VLAN; second handles everything else.
register_serializer_resolver(
lambda model, prefix='': first if model is VLAN else None
)
register_serializer_resolver(lambda model, prefix='': second)
self.assertIs(get_serializer_for_model(VLAN), first)
self.assertIs(get_serializer_for_model(Site), second)
def test_resolver_receives_prefix(self):
from netbox.plugins import register_serializer_resolver
from utilities.api import get_serializer_for_model
seen = {}
def resolver(model, prefix=''):
seen['prefix'] = prefix
return object()
register_serializer_resolver(resolver)
get_serializer_for_model(Site, prefix='Nested')
self.assertEqual(seen['prefix'], 'Nested')
def test_register_rejects_non_callable(self):
from netbox.plugins import register_serializer_resolver
with self.assertRaises(TypeError):
register_serializer_resolver('not a callable')
class APITrailingSlashTestCase(APITestCase):
"""
Verify behavior for REST API requests sent to a URL without a trailing slash.