From 97be961df7596e91f7fa48558a165faade4e673d Mon Sep 17 00:00:00 2001 From: Arthur Date: Thu, 21 May 2026 08:43:48 -0700 Subject: [PATCH] Scope serializer resolvers per-app and drop default discovery path Address review feedback on #22253: - Registry stores resolvers as {app_label: resolver} dict instead of a flat list, so each app can only register a resolver for its own models. - register_serializer_resolver() now takes (app_label, resolver) and get_serializer_for_model() only consults the resolver registered for the model's own app. - Remove 'serializer_resolver' from DEFAULT_RESOURCE_PATHS so this niche resource is loaded only when a plugin explicitly defines it. The PluginConfig.ready() path imports the configured path directly and registers it under self.label. - Update tests for the new per-app scoping; verify a resolver registered for one app does not affect lookups in another. Co-Authored-By: Claude Opus 4.7 (1M context) --- netbox/netbox/plugins/__init__.py | 10 +++--- netbox/netbox/plugins/registration.py | 14 ++++---- netbox/netbox/registry.py | 2 +- netbox/utilities/api.py | 31 ++++++++-------- netbox/utilities/tests/test_api.py | 51 +++++++++++++++------------ 5 files changed, 59 insertions(+), 49 deletions(-) diff --git a/netbox/netbox/plugins/__init__.py b/netbox/netbox/plugins/__init__.py index a23b38c6c..5f72971eb 100644 --- a/netbox/netbox/plugins/__init__.py +++ b/netbox/netbox/plugins/__init__.py @@ -32,7 +32,6 @@ 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', } @@ -136,9 +135,12 @@ class PluginConfig(AppConfig): if user_preferences := self._load_resource('user_preferences'): register_user_preferences(plugin_name, user_preferences) - # Register serializer resolver (if defined) - if serializer_resolver := self._load_resource('serializer_resolver'): - register_serializer_resolver(serializer_resolver) + # Register serializer resolver (if explicitly defined). No default path is attempted. + if self.serializer_resolver: + register_serializer_resolver( + self.label, + import_string(f"{self.__module__}.{self.serializer_resolver}"), + ) @classmethod def validate(cls, user_config, netbox_version): diff --git a/netbox/netbox/plugins/registration.py b/netbox/netbox/plugins/registration.py index ff1cec8c7..1f6091e8b 100644 --- a/netbox/netbox/plugins/registration.py +++ b/netbox/netbox/plugins/registration.py @@ -85,12 +85,14 @@ def register_user_preferences(plugin_name, preferences): registry['plugins']['preferences'][plugin_name] = preferences -def register_serializer_resolver(resolver): +def register_serializer_resolver(app_label, 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(). + Register a callable that returns a DRF serializer class for a model in + the given app, or None if the resolver does not handle the model. The + resolver is consulted by utilities.api.get_serializer_for_model() before + the default import-path lookup, but only for models belonging to + `app_label`. Plugins (and internal apps) should only register resolvers + for their own models. This is the supported extension point for plugins whose models are generated dynamically (and therefore have no importable serializer at @@ -101,4 +103,4 @@ def register_serializer_resolver(resolver): """ if not callable(resolver): raise TypeError(_("Serializer resolver must be callable")) - registry['serializer_resolvers'].append(resolver) + registry['serializer_resolvers'][app_label] = resolver diff --git a/netbox/netbox/registry.py b/netbox/netbox/registry.py index 3edc8b180..151029ca1 100644 --- a/netbox/netbox/registry.py +++ b/netbox/netbox/registry.py @@ -44,7 +44,7 @@ registry = Registry({ 'plugins': dict(), 'request_processors': list(), 'search': dict(), - 'serializer_resolvers': list(), + 'serializer_resolvers': dict(), 'system_jobs': dict(), 'tables': collections.defaultdict(dict), 'views': collections.defaultdict(dict), diff --git a/netbox/utilities/api.py b/netbox/utilities/api.py index c438ad3dc..4a4719418 100644 --- a/netbox/utilities/api.py +++ b/netbox/utilities/api.py @@ -51,30 +51,31 @@ 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. + A plugin (or internal app) may register a custom resolver for its own + app via netbox.plugins.register_serializer_resolver() to handle + dynamically generated models or to override serializer resolution. If + a resolver is registered for the model's app and returns a Serializer + subclass, that result is used. Otherwise, the default import-path + lookup runs. """ - for resolver in registry['serializer_resolvers']: + app_label, model_name = model._meta.label.split('.') + + if resolver := registry['serializer_resolvers'].get(app_label): try: serializer = resolver(model, prefix=prefix) except Exception: # A buggy resolver must not break serializer lookup for the rest of NetBox. - logger.exception("Serializer resolver %r raised an exception; skipping.", resolver) - continue - if serializer is None: - continue - if not (isinstance(serializer, type) and issubclass(serializer, Serializer)): + logger.exception("Serializer resolver %r raised an exception; falling through to default lookup.", resolver) + serializer = None + if serializer is not None: + if isinstance(serializer, type) and issubclass(serializer, Serializer): + return serializer logger.warning( - "Serializer resolver %r returned %r, which is not a Serializer subclass; skipping.", + "Serializer resolver %r returned %r, which is not a Serializer subclass; " + "falling through to default lookup.", resolver, serializer, ) - continue - return serializer - app_label, model_name = model._meta.label.split('.') serializer_name = f'{app_label}.api.serializers.{prefix}{model_name}Serializer' try: return import_string(serializer_name) diff --git a/netbox/utilities/tests/test_api.py b/netbox/utilities/tests/test_api.py index f41e9505f..76016ceba 100644 --- a/netbox/utilities/tests/test_api.py +++ b/netbox/utilities/tests/test_api.py @@ -9,6 +9,7 @@ from dcim.api.serializers import SiteSerializer from dcim.models import Region, Site from extras.choices import CustomFieldTypeChoices from extras.models import CustomField +from ipam.api.serializers import VLANSerializer from ipam.models import VLAN from netbox.api.serializers import BaseModelSerializer from netbox.config import get_config @@ -490,42 +491,47 @@ class _ResolvedSerializerB(Serializer): class SerializerResolverRegistryTestCase(TestCase): """ - Verify that registered serializer resolvers are consulted before the - default import-path lookup in get_serializer_for_model(). + Verify that a registered serializer resolver is consulted before the + default import-path lookup in get_serializer_for_model(), scoped to + the app for which it was registered. """ def setUp(self): - # Snapshot and clear the resolver list so each test starts from a + # Snapshot and clear the resolver mapping 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']) + self._saved_resolvers = dict(registry['serializer_resolvers']) registry['serializer_resolvers'].clear() def tearDown(self): registry['serializer_resolvers'].clear() - registry['serializer_resolvers'].extend(self._saved_resolvers) + registry['serializer_resolvers'].update(self._saved_resolvers) def test_default_lookup_when_no_resolvers_registered(self): self.assertIs(get_serializer_for_model(Site), SiteSerializer) def test_registered_resolver_overrides_default(self): - register_serializer_resolver(lambda model, prefix='': _ResolvedSerializerA) + register_serializer_resolver('dcim', lambda model, prefix='': _ResolvedSerializerA) self.assertIs(get_serializer_for_model(Site), _ResolvedSerializerA) def test_resolver_returning_none_falls_through_to_default(self): - register_serializer_resolver(lambda model, prefix='': None) + register_serializer_resolver('dcim', lambda model, prefix='': None) self.assertIs(get_serializer_for_model(Site), SiteSerializer) - def test_resolvers_tried_in_registration_order(self): - # First resolver only handles VLAN; second handles everything else. - register_serializer_resolver( - lambda model, prefix='': _ResolvedSerializerA if model is VLAN else None - ) - register_serializer_resolver(lambda model, prefix='': _ResolvedSerializerB) + def test_resolver_scoped_to_registered_app(self): + # A resolver registered for dcim must not affect lookups for other apps (e.g. ipam). + register_serializer_resolver('dcim', lambda model, prefix='': _ResolvedSerializerA) - self.assertIs(get_serializer_for_model(VLAN), _ResolvedSerializerA) - self.assertIs(get_serializer_for_model(Site), _ResolvedSerializerB) + self.assertIs(get_serializer_for_model(Site), _ResolvedSerializerA) + self.assertIs(get_serializer_for_model(VLAN), VLANSerializer) + + def test_per_app_resolvers_are_independent(self): + register_serializer_resolver('dcim', lambda model, prefix='': _ResolvedSerializerA) + register_serializer_resolver('ipam', lambda model, prefix='': _ResolvedSerializerB) + + self.assertIs(get_serializer_for_model(Site), _ResolvedSerializerA) + self.assertIs(get_serializer_for_model(VLAN), _ResolvedSerializerB) def test_resolver_receives_prefix(self): seen = {} @@ -534,27 +540,26 @@ class SerializerResolverRegistryTestCase(TestCase): seen['prefix'] = prefix return _ResolvedSerializerA - register_serializer_resolver(resolver) + register_serializer_resolver('dcim', resolver) get_serializer_for_model(Site, prefix='Nested') self.assertEqual(seen['prefix'], 'Nested') def test_register_rejects_non_callable(self): with self.assertRaises(TypeError): - register_serializer_resolver('not a callable') + register_serializer_resolver('dcim', 'not a callable') - def test_raising_resolver_is_skipped(self): + def test_raising_resolver_falls_through_to_default(self): def broken_resolver(model, prefix=''): raise RuntimeError("intentional failure") - register_serializer_resolver(broken_resolver) - register_serializer_resolver(lambda model, prefix='': _ResolvedSerializerA) + register_serializer_resolver('dcim', broken_resolver) with self.assertLogs('netbox.utilities.api', level='ERROR'): - self.assertIs(get_serializer_for_model(Site), _ResolvedSerializerA) + self.assertIs(get_serializer_for_model(Site), SiteSerializer) - def test_resolver_returning_non_serializer_is_skipped(self): - register_serializer_resolver(lambda model, prefix='': object()) + def test_resolver_returning_non_serializer_falls_through_to_default(self): + register_serializer_resolver('dcim', lambda model, prefix='': object()) with self.assertLogs('netbox.utilities.api', level='WARNING'): self.assertIs(get_serializer_for_model(Site), SiteSerializer)