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) <noreply@anthropic.com>
This commit is contained in:
Arthur
Jeremy Stretch
parent
ff26cbd521
commit
97be961df7
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in New Issue