Closes #23547: Refactor the netbox.utils module (#23549)

docs/development/application-registry.md

@@ -35,7 +35,7 @@ This key lists all models which have been registered in NetBox which are not des
 
 ### `pk_allocators`
 
-A list of callables which may supply primary keys for new objects, consulted in order. Primary key allocators are registered using the `register_pk_allocator()` decorator in `netbox.utils`. (See the [plugin development documentation](../plugins/development/models.md#primary-key-allocators) for details.)
+A list of callables which may supply primary keys for new objects, consulted in order. Primary key allocators are registered using the `register_pk_allocator()` decorator in `netbox.models`. (See the [plugin development documentation](../plugins/development/models.md#primary-key-allocators) for details.)
 
 ### `plugins`
 
@@ -43,7 +43,7 @@ This store maintains all registered items for plugins, such as navigation menus,
 
 ### `request_processors`
 
-A list of context managers to invoke when processing a request e.g. in middleware or when executing a background job. Request processors can be registered with the `@register_request_processor` decorator.
+A list of context managers to invoke when processing a request e.g. in middleware or when executing a background job. Request processors can be registered with the `@register_request_processor` decorator in `utilities.request`.
 
 ### `search`
 

docs/plugins/development/models.md

@@ -214,10 +214,10 @@ The caller is responsible for calling the method only once every related object
 
 !!! info "This feature was introduced in NetBox v4.8."
 
-By default, the database assigns each new object's primary key from its table's sequence. A plugin can instead supply primary keys by registering a primary key allocator with the `register_pk_allocator()` decorator from `netbox.utils`. An allocator is a callable which accepts a model class, a count, and the alias of the database to which the new objects will be written. It returns either a list of exactly that many primary keys for the model, or `None` to defer to the next allocator.
+By default, the database assigns each new object's primary key from its table's sequence. A plugin can instead supply primary keys by registering a primary key allocator with the `register_pk_allocator()` decorator from `netbox.models`. An allocator is a callable which accepts a model class, a count, and the alias of the database to which the new objects will be written. It returns either a list of exactly that many primary keys for the model, or `None` to defer to the next allocator.
 
 ```python
-from netbox.utils import register_pk_allocator
+from netbox.models import register_pk_allocator
 
 @register_pk_allocator
 def my_allocator(model, count, using):
@@ -249,10 +249,10 @@ A few things to keep in mind when writing an allocator:
 
 ### Inserting Rows Outside the ORM
 
-Code which inserts rows directly, for example using raw SQL (including `RunSQL` migration operations) or PostgreSQL's `COPY`, bypasses the ORM and with it any registered allocators. Such code should call `allocate_pks()` from `netbox.utils` to obtain primary keys for the new rows, and insert them explicitly. This function returns a list of primary keys from the first allocator which supplies them, or `None` if the database should assign them. As with `save()`, allocators receive the model which owns the primary key column, and are consulted only if it's an auto field. Pass the alias of the database being written to as `using`; if it's omitted, the database router's choice for writing the model is assumed.
+Code which inserts rows directly, for example using raw SQL (including `RunSQL` migration operations) or PostgreSQL's `COPY`, bypasses the ORM and with it any registered allocators. Such code should call `allocate_pks()` from `netbox.models` to obtain primary keys for the new rows, and insert them explicitly. This function returns a list of primary keys from the first allocator which supplies them, or `None` if the database should assign them. As with `save()`, allocators receive the model which owns the primary key column, and are consulted only if it's an auto field. Pass the alias of the database being written to as `using`; if it's omitted, the database router's choice for writing the model is assumed.
 
 ```python
-from netbox.utils import allocate_pks
+from netbox.models import allocate_pks
 
 pks = allocate_pks(MyModel, count=len(rows), using=connection.alias)
 if pks is None:

netbox/core/api/serializers_/data.py

@@ -2,7 +2,7 @@ from core.choices import *
 from core.models import DataFile, DataSource
 from netbox.api.fields import ChoiceField, RelatedObjectCountField
 from netbox.api.serializers import NetBoxModelSerializer, PrimaryModelSerializer
-from netbox.utils import get_data_backend_choices
+from netbox.data_backends import get_data_backend_choices
 
 __all__ = (
     'DataFileSerializer',

netbox/core/data_backends.py

@@ -9,8 +9,7 @@ from django import forms
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.translation import gettext as _
 
-from netbox.data_backends import DataBackend
-from netbox.utils import register_data_backend
+from netbox.data_backends import DataBackend, register_data_backend
 from utilities.constants import HTTP_PROXY_SUPPORTED_SCHEMAS, HTTP_PROXY_SUPPORTED_SOCK_SCHEMAS
 from utilities.proxy import resolve_proxies
 from utilities.socks import ProxyPoolManager

netbox/core/filtersets.py

@@ -3,8 +3,8 @@ from django.contrib.contenttypes.models import ContentType
 from django.db.models import Q
 from django.utils.translation import gettext as _
 
+from netbox.data_backends import get_data_backend_choices
 from netbox.filtersets import BaseFilterSet, ChangeLoggedModelFilterSet, PrimaryModelFilterSet
-from netbox.utils import get_data_backend_choices
 from users.models import User
 from utilities.filters import MultiValueContentTypeFilter
 from utilities.filtersets import register_filterset

netbox/core/forms/bulk_edit.py

@@ -3,8 +3,8 @@ from django.utils.translation import gettext_lazy as _
 
 from core.choices import JobIntervalChoices
 from core.models import *
+from netbox.data_backends import get_data_backend_choices
 from netbox.forms import PrimaryModelBulkEditForm
-from netbox.utils import get_data_backend_choices
 from utilities.forms.fields import ChoiceField
 from utilities.forms.rendering import FieldSet
 from utilities.forms.widgets import BulkEditNullBooleanSelect

netbox/core/forms/filtersets.py

@@ -3,9 +3,9 @@ from django.utils.translation import gettext_lazy as _
 
 from core.choices import *
 from core.models import *
+from netbox.data_backends import get_data_backend_choices
 from netbox.forms import NetBoxModelFilterSetForm, PrimaryModelFilterSetForm
 from netbox.forms.mixins import SavedFiltersMixin
-from netbox.utils import get_data_backend_choices
 from users.models import User
 from utilities.forms import BOOLEAN_WITH_BLANK_CHOICES, FilterForm, add_blank_choice
 from utilities.forms.fields import (

netbox/core/forms/model_forms.py

@@ -9,9 +9,9 @@ from django.utils.translation import gettext_lazy as _
 from core.forms.mixins import SyncedDataMixin
 from core.models import *
 from netbox.config import PARAMS, get_config
+from netbox.data_backends import get_data_backend_choices
 from netbox.forms import NetBoxModelForm, PrimaryModelForm
 from netbox.registry import registry
-from netbox.utils import get_data_backend_choices
 from utilities.forms import get_field_value
 from utilities.forms.fields import ChoiceField, JSONField
 from utilities.forms.rendering import FieldSet

netbox/netbox/context_managers.py

@@ -3,7 +3,7 @@ from contextlib import contextmanager
 
 from extras.events import flush_events
 from netbox.context import current_request, events_queue, query_cache
-from netbox.utils import register_request_processor
+from utilities.request import register_request_processor
 
 
 @register_request_processor

netbox/netbox/data_backends.py

@@ -1,8 +1,13 @@
 from contextlib import contextmanager
 from urllib.parse import urlparse
 
+from netbox.registry import registry
+from utilities.choices import Choice
+
 __all__ = (
     'DataBackend',
+    'get_data_backend_choices',
+    'register_data_backend',
 )
 
 
@@ -51,3 +56,23 @@ class DataBackend:
         3. Performs any necessary cleanup
         """
         raise NotImplementedError()
+
+
+def get_data_backend_choices():
+    return [
+        (None, '---------'),
+        *[
+            Choice(name, cls.label) for name, cls in registry['data_backends'].items()
+        ]
+    ]
+
+
+def register_data_backend():
+    """
+    Decorator for registering a DataBackend class.
+    """
+    def _wrapper(cls):
+        registry['data_backends'][cls.name] = cls
+        return cls
+
+    return _wrapper

netbox/netbox/models/__init__.py

@@ -11,6 +11,7 @@ from netbox.models.features import *
 from netbox.models.ltree import LtreeManager, LtreeModel, SortPathField
 from netbox.models.mixins import OwnerMixin
 from utilities.mptt import TreeManager
+from utilities.pk_allocation import allocate_pks, register_pk_allocator
 from utilities.querysets import RestrictedQuerySet
 
 __all__ = (
@@ -22,6 +23,8 @@ __all__ = (
     'NetBoxModel',
     'OrganizationalModel',
     'PrimaryModel',
+    'allocate_pks',
+    'register_pk_allocator',
 )
 
 

netbox/netbox/plugins/__init__.py

@@ -7,10 +7,10 @@ from django.utils.module_loading import import_string
 from packaging import version
 
 from core.exceptions import IncompatiblePluginError
+from netbox.data_backends import register_data_backend
 from netbox.event_rules import register_event_rule_action
 from netbox.registry import registry
 from netbox.search import register_search
-from netbox.utils import register_data_backend
 
 from .navigation import *
 from .registration import *

netbox/netbox/tests/test_pk_allocators.py

@@ -12,8 +12,8 @@ from account.models import UserToken
 from core.models import ObjectType
 from dcim.models import Manufacturer, Site
 from extras.models import CachedValue, Tag
+from netbox.models import allocate_pks, register_pk_allocator
 from netbox.registry import registry
-from netbox.utils import allocate_pks, register_pk_allocator
 from tenancy.models import Tenant
 from users.models import Token, User
 from utilities import pk_allocation

netbox/netbox/tests/test_utils.py

@@ -0,0 +1,37 @@
+from django.test import SimpleTestCase
+
+from netbox import utils
+from netbox.data_backends import get_data_backend_choices, register_data_backend
+from utilities.request import register_request_processor
+
+# TODO: Remove in NetBox v5.0
+RELOCATED_FUNCTIONS = (
+    get_data_backend_choices,
+    register_data_backend,
+    register_request_processor,
+)
+
+
+# TODO: Remove in NetBox v5.0
+class RelocatedFunctionsTestCase(SimpleTestCase):
+    """
+    Verify that functions relocated from netbox.utils remain importable from it, with a deprecation warning.
+    (Ref: #23547)
+    """
+    def test_relocated_functions(self):
+        for func in RELOCATED_FUNCTIONS:
+            name = func.__name__
+            with self.subTest(name=name):
+                with self.assertWarnsRegex(DeprecationWarning, rf'{name}\(\) from netbox\.utils'):
+                    self.assertIs(getattr(utils, name), func)
+
+    def test_wildcard_import(self):
+        namespace = {}
+        with self.assertWarns(DeprecationWarning):
+            exec('from netbox.utils import *', namespace)
+        for func in RELOCATED_FUNCTIONS:
+            self.assertIs(namespace[func.__name__], func)
+
+    def test_unknown_attribute(self):
+        with self.assertRaises(AttributeError):
+            utils.nonexistent_function

netbox/netbox/utils.py

@@ -1,37 +1,32 @@
-from django.db import router
-from django.db.models.fields import AutoFieldMixin
+import warnings
+from importlib import import_module
 
 from netbox.registry import registry
-from utilities.choices import Choice
+
+# Functions which have been relocated from this module, mapped to their new modules
+# TODO: Remove in NetBox v5.0
+RELOCATED_FUNCTIONS = {
+    'get_data_backend_choices': 'netbox.data_backends',
+    'register_data_backend': 'netbox.data_backends',
+    'register_request_processor': 'utilities.request',
+}
 
 __all__ = (
-    'allocate_pks',
-    'get_data_backend_choices',
-    'register_data_backend',
     'register_model_feature',
-    'register_pk_allocator',
-    'register_request_processor',
+    *RELOCATED_FUNCTIONS,  # TODO: Remove in NetBox v5.0
 )
 
 
-def get_data_backend_choices():
-    return [
-        (None, '---------'),
-        *[
-            Choice(name, cls.label) for name, cls in registry['data_backends'].items()
-        ]
-    ]
-
-
-def register_data_backend():
-    """
-    Decorator for registering a DataBackend class.
-    """
-    def _wrapper(cls):
-        registry['data_backends'][cls.name] = cls
-        return cls
-
-    return _wrapper
+def __getattr__(name):
+    if module := RELOCATED_FUNCTIONS.get(name):
+        warnings.warn(
+            f"Importing {name}() from netbox.utils is deprecated and will be removed in NetBox v5.0. Import it from "
+            f"{module} instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return getattr(import_module(module), name)
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 
 def register_model_feature(name, func=None):
@@ -61,62 +56,3 @@ def register_model_feature(name, func=None):
     if func is None:
         return decorator
     return decorator(func)
-
-
-def register_request_processor(func):
-    """
-    Decorator for registering a request processor.
-    """
-    registry['request_processors'].append(func)
-
-    return func
-
-
-def register_pk_allocator(func):
-    """
-    Decorator for registering a primary key allocator: a callable which may supply primary keys for new objects.
-
-    The callable must accept a model class, a count, and the alias of the database to which the new objects will be
-    written (`using`), and return either a list of exactly `count` primary keys for that model, or None to defer to the
-    next allocator (and ultimately to the database). Allocators are consulted in the order in which they were
-    registered. Registering an allocator which is already registered has no effect.
-
-        @register_pk_allocator
-        def my_allocator(model, count, using):
-            ...
-    """
-    if func not in registry['pk_allocators']:
-        registry['pk_allocators'].append(func)
-
-    return func
-
-
-def allocate_pks(model, count=1, using=None):
-    """
-    Return a list of `count` primary keys for new instances of `model` from the first registered allocator which
-    supplies them, or None if the database should assign them. `using` is the alias of the database to which the new
-    instances will be written; if omitted, the database router's choice for writing `model` is assumed. Code which
-    inserts rows outside the ORM (e.g. using raw SQL or COPY) can call this to obtain primary keys in the same manner
-    as Model.save() and QuerySet.bulk_create().
-    """
-    if using is None:
-        using = router.db_for_write(model)
-    # Allocators receive the model which owns the primary key column (the concrete parent, for a proxy or multi-table
-    # inheritance child), and only if it's an auto field
-    pk = model._meta.pk
-    while pk.remote_field and pk.remote_field.parent_link:
-        pk = pk.target_field
-    if not isinstance(pk, AutoFieldMixin):
-        return None
-    model = pk.model
-    for allocator in registry['pk_allocators']:
-        if (pks := allocator(model, count, using)) is not None:
-            pks = list(pks)
-            if len(pks) != count:
-                name = getattr(allocator, '__qualname__', allocator)
-                raise ValueError(
-                    f'Primary key allocator {name} returned {len(pks)} primary keys for {model._meta.label} '
-                    f'(expected {count}).'
-                )
-            return pks
-    return None

netbox/utilities/pk_allocation.py

@@ -1,12 +1,13 @@
-from django.db import models
+from django.db import models, router
 from django.db.models.fields import AutoFieldMixin
 
 from netbox.registry import registry
-from netbox.utils import allocate_pks
 
 __all__ = (
+    'allocate_pks',
     'bulk_create_with_pk_allocation',
     'prepare_for_bulk_create_with_pk_allocation',
+    'register_pk_allocator',
     'save_table_with_pk_allocation',
 )
 
@@ -17,6 +18,56 @@ _prepare_for_bulk_create = models.QuerySet._prepare_for_bulk_create
 _save_table = models.Model._save_table
 
 
+def register_pk_allocator(func):
+    """
+    Decorator for registering a primary key allocator: a callable which may supply primary keys for new objects.
+
+    The callable must accept a model class, a count, and the alias of the database to which the new objects will be
+    written (`using`), and return either a list of exactly `count` primary keys for that model, or None to defer to the
+    next allocator (and ultimately to the database). Allocators are consulted in the order in which they were
+    registered. Registering an allocator which is already registered has no effect.
+
+        @register_pk_allocator
+        def my_allocator(model, count, using):
+            ...
+    """
+    if func not in registry['pk_allocators']:
+        registry['pk_allocators'].append(func)
+
+    return func
+
+
+def allocate_pks(model, count=1, using=None):
+    """
+    Return a list of `count` primary keys for new instances of `model` from the first registered allocator which
+    supplies them, or None if the database should assign them. `using` is the alias of the database to which the new
+    instances will be written; if omitted, the database router's choice for writing `model` is assumed. Code which
+    inserts rows outside the ORM (e.g. using raw SQL or COPY) can call this to obtain primary keys in the same manner
+    as Model.save() and QuerySet.bulk_create().
+    """
+    if using is None:
+        using = router.db_for_write(model)
+    # Allocators receive the model which owns the primary key column (the concrete parent, for a proxy or multi-table
+    # inheritance child), and only if it's an auto field
+    pk = model._meta.pk
+    while pk.remote_field and pk.remote_field.parent_link:
+        pk = pk.target_field
+    if not isinstance(pk, AutoFieldMixin):
+        return None
+    model = pk.model
+    for allocator in registry['pk_allocators']:
+        if (pks := allocator(model, count, using)) is not None:
+            pks = list(pks)
+            if len(pks) != count:
+                name = getattr(allocator, '__qualname__', allocator)
+                raise ValueError(
+                    f'Primary key allocator {name} returned {len(pks)} primary keys for {model._meta.label} '
+                    f'(expected {count}).'
+                )
+            return pks
+    return None
+
+
 def _can_allocate_pks(model):
     """
     Return True if primary keys for new instances of the model may be obtained from the registered allocators.

netbox/utilities/request.py

@@ -18,6 +18,7 @@ __all__ = (
     'copy_safe_request',
     'get_client_ip',
     'get_safe_request_context',
+    'register_request_processor',
     'safe_for_redirect',
 )
 
@@ -132,6 +133,15 @@ def safe_for_redirect(url):
     return url_has_allowed_host_and_scheme(url, allowed_hosts=None)
 
 
+def register_request_processor(func):
+    """
+    Decorator for registering a request processor.
+    """
+    registry['request_processors'].append(func)
+
+    return func
+
+
 @contextmanager
 def apply_request_processors(request):
     """