Просмотр исходного кода

allow plugins to override get_model_serializer

Arthur 2 недель назад
Родитель
Сommit
390c697337

+ 7 - 0
netbox/netbox/plugins/__init__.py

@@ -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):
 

+ 20 - 0
netbox/netbox/plugins/registration.py

@@ -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)

+ 1 - 0
netbox/netbox/registry.py

@@ -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),

+ 12 - 0
netbox/utilities/api.py

@@ -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:

+ 81 - 0
netbox/utilities/tests/test_api.py

@@ -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.