148 Allow NetBox Scripts to run when a Branch is selected

netbox_branching/__init__.py

@@ -11,7 +11,7 @@ class AppConfig(PluginConfig):
     description = 'A git-like branching implementation for NetBox'
     version = '0.5.2'
     base_url = 'branching'
-    min_version = '4.1'
+    min_version = '4.1.9'
     middleware = [
         'netbox_branching.middleware.BranchMiddleware'
     ]

netbox_branching/middleware.py

@@ -1,14 +1,8 @@
-from django.contrib import messages
 from django.core.exceptions import ObjectDoesNotExist
 from django.http import HttpResponseBadRequest
-from django.urls import reverse
 
-from utilities.api import is_api_request
-
-from .choices import BranchStatusChoices
-from .constants import COOKIE_NAME, BRANCH_HEADER, QUERY_PARAM
-from .models import Branch
-from .utilities import activate_branch, is_api_request
+from .constants import COOKIE_NAME, QUERY_PARAM
+from .utilities import activate_branch, is_api_request, get_active_branch
 
 __all__ = (
     'BranchMiddleware',
@@ -24,12 +18,11 @@ def __call__(self, request):
 
         # Set/clear the active Branch on the request
         try:
-            branch = self.get_active_branch(request)
+            branch = get_active_branch(request)
         except ObjectDoesNotExist:
             return HttpResponseBadRequest("Invalid branch identifier")
 
-        with activate_branch(branch):
-            response = self.get_response(request)
+        response = self.get_response(request)
 
         # Set/clear the branch cookie (for non-API requests)
         if not is_api_request(request):
@@ -39,34 +32,3 @@ def __call__(self, request):
                 response.delete_cookie(COOKIE_NAME)
 
         return response
-
-    @staticmethod
-    def get_active_branch(request):
-        """
-        Return the active Branch (if any).
-        """
-        # The active Branch may be specified by HTTP header for REST & GraphQL API requests.
-        if is_api_request(request) and BRANCH_HEADER in request.headers:
-            branch = Branch.objects.get(schema_id=request.headers.get(BRANCH_HEADER))
-            if not branch.ready:
-                return HttpResponseBadRequest(f"Branch {branch} is not ready for use (status: {branch.status})")
-            return branch
-
-        # Branch activated/deactivated by URL query parameter
-        elif QUERY_PARAM in request.GET:
-            if schema_id := request.GET.get(QUERY_PARAM):
-                branch = Branch.objects.get(schema_id=schema_id)
-                if branch.ready:
-                    messages.success(request, f"Activated branch {branch}")
-                    return branch
-                else:
-                    messages.error(request, f"Branch {branch} is not ready for use (status: {branch.status})")
-                    return None
-            else:
-                messages.success(request, f"Deactivated branch")
-                request.COOKIES.pop(COOKIE_NAME, None)  # Delete cookie if set
-                return None
-
-        # Branch set by cookie
-        elif schema_id := request.COOKIES.get(COOKIE_NAME):
-            return Branch.objects.filter(schema_id=schema_id, status=BranchStatusChoices.READY).first()

netbox_branching/template_content.py

@@ -1,13 +1,15 @@
 from django.contrib.contenttypes.models import ContentType
-from netbox.plugins import PluginTemplateExtension
 
+from netbox.plugins import PluginTemplateExtension
 from .choices import BranchStatusChoices
 from .contextvars import active_branch
 from .models import Branch, ChangeDiff
 
 __all__ = (
     'BranchNotification',
     'BranchSelector',
+    'ScriptNotification',
+    'ShareButton',
     'template_extensions',
 )
 
@@ -34,6 +36,7 @@ class BranchNotification(PluginTemplateExtension):
     def alerts(self):
         if not (instance := self.context['object']):
             return ''
+
         ct = ContentType.objects.get_for_model(instance)
         relevant_changes = ChangeDiff.objects.filter(
             object_type=ct,
@@ -51,4 +54,18 @@ def alerts(self):
         })
 
 
-template_extensions = [BranchSelector, ShareButton, BranchNotification]
+class ScriptNotification(PluginTemplateExtension):
+    models = ['extras.script']
+
+    def alerts(self):
+        return self.render('netbox_branching/inc/script_alert.html', extra_context={
+            'active_branch': active_branch.get(),
+        })
+
+
+template_extensions = (
+    BranchSelector,
+    BranchNotification,
+    ScriptNotification,
+    ShareButton,
+)

netbox_branching/templates/netbox_branching/inc/script_alert.html

@@ -0,0 +1,8 @@
+{% load i18n %}
+{% if active_branch %}
+  <div class="alert alert-warning" role="alert">
+    <i class="mdi mdi-alert"></i>
+    {% trans "This script will be run in branch" %}
+      <a href="{{ active_branch.get_absolute_url }}">{{ active_branch.name }}</a>
+  </div>
+{% endif %}

netbox_branching/utilities.py

@@ -1,23 +1,29 @@
 import datetime
 import logging
 from collections import defaultdict
-from contextlib import contextmanager
+from contextlib import contextmanager, nullcontext
 from dataclasses import dataclass
 
+from django.contrib import messages
 from django.db.models import ForeignKey, ManyToManyField
+from django.http import HttpResponseBadRequest
 from django.urls import reverse
 
 from netbox.plugins import get_plugin_config
 from netbox.registry import registry
-from .constants import EXEMPT_MODELS, INCLUDE_MODELS
+from netbox.utils import register_request_processor
+from .choices import BranchStatusChoices
+from .constants import BRANCH_HEADER, COOKIE_NAME, EXEMPT_MODELS, INCLUDE_MODELS, QUERY_PARAM
 from .contextvars import active_branch
 
 __all__ = (
     'ChangeSummary',
     'DynamicSchemaDict',
     'ListHandler',
+    'ActiveBranchContextManager',
     'activate_branch',
     'deactivate_branch',
+    'get_active_branch',
     'get_branchable_object_types',
     'get_tables_to_replicate',
     'is_api_request',
@@ -209,4 +215,49 @@ def is_api_request(request):
     """
     Returns True if the given request is a REST or GraphQL API request.
     """
+    if not hasattr(request, 'path_info'):
+        return False
+
     return request.path_info.startswith(reverse('api-root')) or request.path_info.startswith(reverse('graphql'))
+
+
+def get_active_branch(request):
+    """
+    Return the active Branch (if any).
+    """
+    # The active Branch may be specified by HTTP header for REST & GraphQL API requests.
+    from .models import Branch
+    if is_api_request(request) and BRANCH_HEADER in request.headers:
+        branch = Branch.objects.get(schema_id=request.headers.get(BRANCH_HEADER))
+        if not branch.ready:
+            return HttpResponseBadRequest(f"Branch {branch} is not ready for use (status: {branch.status})")
+        return branch
+
+    # Branch activated/deactivated by URL query parameter
+    elif QUERY_PARAM in request.GET:
+        if schema_id := request.GET.get(QUERY_PARAM):
+            branch = Branch.objects.get(schema_id=schema_id)
+            if branch.ready:
+                messages.success(request, f"Activated branch {branch}")
+                return branch
+            else:
+                messages.error(request, f"Branch {branch} is not ready for use (status: {branch.status})")
+                return None
+        else:
+            messages.success(request, f"Deactivated branch")
+            request.COOKIES.pop(COOKIE_NAME, None)  # Delete cookie if set
+            return None
+
+    # Branch set by cookie
+    elif schema_id := request.COOKIES.get(COOKIE_NAME):
+        return Branch.objects.filter(schema_id=schema_id, status=BranchStatusChoices.READY).first()
+
+
+@register_request_processor
+def ActiveBranchContextManager(request):
+    """
+    Activate a branch if indicated by the request.
+    """
+    if branch := get_active_branch(request):
+        return activate_branch(branch)
+    return nullcontext()