Django Architecture

Project Structure

plugin_name/
├── __init__.py          # PluginConfig with adapter validation in ready()
├── models/              # Active Record models (aggregate roots)
│   ├── __init__.py
│   └── order.py
├── services/            # Cross-model orchestration, workflows
│   ├── __init__.py
│   └── provisioning.py
├── adapters.py          # Ports (Protocol) + Adapters + Loaders
├── api/                 # DRF serializers, viewsets, urls
│   ├── serializers.py
│   ├── viewsets.py
│   └── urls.py
├── views/               # Django/NetBox UI views (thin)
├── forms/               # Django forms (UI rendering only)
├── tables/              # NetBox tables
├── filtersets.py        # Django-filter querysets
├── choices/             # Enums (TextChoices)
├── validators.py        # Field validators with normalization
├── jobs.py              # Async/background jobs
├── navigation.py        # NetBox navigation config
├── templates/
├── migrations/
└── tests/

Model Patterns

Field Design

from django.db import models
from netbox.models import NetBoxModel

class ServiceOrder(NetBoxModel):
    """Aggregate root for service provisioning."""

    # Core data
    name = models.CharField(max_length=100)
    order_type = models.CharField(max_length=50, choices=OrderTypeChoices)
    status = models.CharField(
        max_length=30,
        choices=StatusChoices,
        default=StatusChoices.DRAFT,
    )

    # Cross-plugin reference (ID, not ForeignKey)
    circuit_id = models.PositiveBigIntegerField(null=True, blank=True)

    # Timestamps
    submitted_at = models.DateTimeField(null=True, blank=True)
    completed_at = models.DateTimeField(null=True, blank=True)

Cross-plugin references: Store integer IDs, never ForeignKey across plugin boundaries. Resolve via adapters at runtime.

Row-Level Business Logic

Model methods encapsulate state transitions and business rules. They change state but do not call save() — the caller decides when to persist. Maintain query/command separation: queries return data with no side effects (can_submit(), is_overdue()), commands change state and return None (submit(), complete()).

class ServiceOrder(NetBoxModel):

    def submit(self) -> None:
        """Command: transition from DRAFT to SUBMITTED."""
        if self.status != StatusChoices.DRAFT:
            raise ValidationError(f"Cannot submit order in status {self.status}")
        self.status = StatusChoices.SUBMITTED
        self.submitted_at = timezone.now()

    def can_submit(self) -> bool:
        """Query: check if submission is allowed."""
        return self.status == StatusChoices.DRAFT

Validation

# Field-level: validators.py (with normalization)
def validate_circuit_reference(value: str) -> str:
    """Validate and normalize circuit reference format."""
    normalized = value.strip().upper()
    if not re.match(r"^CIR-\d{6}$", normalized):
        raise ValidationError(f"Invalid circuit reference: {value}")
    return normalized

# Model-level: cross-field validation in clean()
class ServiceOrder(NetBoxModel):
    def clean(self):
        super().clean()
        if self.order_type == OrderTypeChoices.UPGRADE and not self.circuit_id:
            raise ValidationError(
                {"circuit_id": "Circuit required for upgrade orders."}
            )

Custom Managers and QuerySets

class ServiceOrderQuerySet(models.QuerySet):
    def active(self):
        return self.exclude(status=StatusChoices.CANCELLED)

    def pending_provisioning(self):
        return self.filter(status=StatusChoices.SUBMITTED)

class ServiceOrderManager(models.Manager):
    def get_queryset(self):
        return ServiceOrderQuerySet(self.model, using=self._db)

    def create_draft(self, **kwargs) -> "ServiceOrder":
        """Factory method for creating orders in DRAFT status."""
        return self.create(status=StatusChoices.DRAFT, **kwargs)

Enums with TextChoices

class StatusChoices(TextChoices):
    DRAFT = "draft", "Draft"
    SUBMITTED = "submitted", "Submitted"
    PROVISIONING = "provisioning", "Provisioning"
    COMPLETED = "completed", "Completed"

Aggregate Root with OneToOneField

Use OneToOneField for polymorphic sub-types sharing a common base:

class ServiceOrder(NetBoxModel):
    """Base aggregate root."""
    order_type = models.CharField(max_length=50, choices=OrderTypeChoices)

class NewCircuitOrder(models.Model):
    """Sub-type specific fields for new circuit orders."""
    order = models.OneToOneField(
        ServiceOrder,
        on_delete=models.CASCADE,
        related_name="%(class)s",
    )
    bandwidth = models.CharField(max_length=20)
    location_a = models.CharField(max_length=200)
    location_b = models.CharField(max_length=200)

Service Layer

When to Use Services

• Cross-model coordination (order + line items + external provisioning)
• Multi-instance logic (operating on multiple instances of the same model)
• External adapter calls (anything behind a Protocol)
• Multi-step workflows needing transaction.atomic
• Business rules that span multiple models or plugins

What Stays on Models

• Single-instance state transitions (order.submit())
• Single-instance queries (order.can_submit())
• Row-level validation (clean())
• Custom QuerySet filters for the model's own table (.active(), .pending())

Template Method with Factory

from abc import ABC, abstractmethod
from django.db import transaction

class OrderProvisioningService(ABC):
    """Abstract workflow: validate -> create -> provision."""

    def __init__(self, **adapters):
        for name, adapter in adapters.items():
            setattr(self, name, adapter)

    @classmethod
    def for_order_type(cls, order_type: str, **adapters):
        """Factory: resolve correct service subclass from order type."""
        registry = {
            OrderTypeChoices.NEW_CIRCUIT: NewCircuitProvisioningService,
            OrderTypeChoices.UPGRADE: UpgradeProvisioningService,
        }
        service_cls = registry.get(order_type)
        if service_cls is None:
            raise ValueError(f"Unknown order type: {order_type}")
        return service_cls(**adapters)

    @transaction.atomic
    def create_order(self, data: dict) -> ServiceOrder:
        """Template method defining the workflow skeleton."""
        self.validate_references(data)
        order = ServiceOrder.objects.create_draft(
            name=data["name"], order_type=self.order_type,
        )
        self.create_sub_type_record(order, data)
        return order

    @abstractmethod
    def validate_references(self, data: dict) -> None: ...

    @abstractmethod
    def create_sub_type_record(self, order: ServiceOrder, data: dict) -> None: ...

class NewCircuitProvisioningService(OrderProvisioningService):
    order_type = OrderTypeChoices.NEW_CIRCUIT

    def validate_references(self, data: dict) -> None:
        if not self.location_adapter.location_exists(data["location_a_id"]):
            raise ValidationError("Location A not found")

    def create_sub_type_record(self, order: ServiceOrder, data: dict) -> None:
        NewCircuitOrder.objects.create(
            order=order, bandwidth=data["bandwidth"],
            location_a=data["location_a"], location_b=data["location_b"],
        )

Batch Operations for Background Jobs

class ProvisioningSync:
    @classmethod
    def sync_all_pending(cls, **adapters) -> int:
        synced = 0
        for order in ServiceOrder.objects.pending_provisioning():
            service = OrderProvisioningService.for_order_type(
                order.order_type, **adapters
            )
            service.provision(order)
            synced += 1
        return synced

Adapter Pattern with Settings-Based DI

Protocol as Port

Define what the domain needs. Keep ports focused — separate read (query) from write (command) when appropriate.

from typing import Protocol, runtime_checkable

@runtime_checkable
class CircuitAdapter(Protocol):
    """Port for circuit operations across plugin boundaries."""

    def circuit_exists(self, circuit_id: int) -> bool: ...
    def get_circuit_display(self, circuit_id: int) -> str | None: ...
    def create_circuit(self, data: dict) -> int: ...

Production Adapter (Django ORM)

Inline imports in adapters are acceptable — they isolate the dependency to the adapter boundary.

class DjangoCircuitAdapter:
    def circuit_exists(self, circuit_id: int) -> bool:
        from circuits.models import Circuit
        return Circuit.objects.filter(pk=circuit_id).exists()

    def get_circuit_display(self, circuit_id: int) -> str | None:
        from circuits.models import Circuit
        try:
            return str(Circuit.objects.get(pk=circuit_id))
        except Circuit.DoesNotExist:
            return None

In-Memory Adapter (Test Double)

class InMemoryCircuitAdapter:
    """Stateful test double — no mocking required."""
    _circuits: dict[int, dict] = {}
    _next_id: int = 1

    @classmethod
    def reset(cls) -> None:
        cls._circuits = {}
        cls._next_id = 1

    def circuit_exists(self, circuit_id: int) -> bool:
        return circuit_id in self._circuits

    def get_circuit_display(self, circuit_id: int) -> str | None:
        circuit = self._circuits.get(circuit_id)
        return circuit["name"] if circuit else None

Settings-Based Loading

# adapters.py — loader functions

from django.conf import settings
from django.utils.module_loading import import_string

def get_circuit_adapter() -> CircuitAdapter:
    """Load the circuit adapter from plugin settings."""
    config = settings.PLUGINS_CONFIG.get("noa_service_order", {})
    adapter_path = config.get(
        "CIRCUIT_ADAPTER",
        "noa_service_order.adapters.DjangoCircuitAdapter",
    )
    adapter_cls = import_string(adapter_path)
    return adapter_cls()

Configuration and Startup Validation

# NetBox configuration.py
PLUGINS_CONFIG = {
    "noa_service_order": {
        "CIRCUIT_ADAPTER": "noa_service_order.adapters.DjangoCircuitAdapter",
    },
}

Validate required adapters are importable at startup in PluginConfig.ready():

class NServiceOrderConfig(PluginConfig):
    name = "noa_service_order"
    required_adapters = ["CIRCUIT_ADAPTER", "LOCATION_ADAPTER"]

    def ready(self):
        super().ready()
        config = self._get_plugin_config()
        for key in self.required_adapters:
            path = config.get(key)
            if path:
                try:
                    import_string(path)
                except ImportError as e:
                    raise ImportError(f"Cannot import {key}={path}: {e}")

Serializer and ViewSet Patterns

Pass adapters via get_serializer_context() for computed fields. Use SerializerMethodField for adapter-backed output enrichment. Override to_internal_value() / to_representation() for complex bidirectional conversion.

class ServiceOrderSerializer(NetBoxModelSerializer):
    circuit_id = serializers.IntegerField(required=False, allow_null=True)
    circuit_display = serializers.SerializerMethodField()

    class Meta:
        model = ServiceOrder
        fields = ["id", "name", "order_type", "status",
                  "circuit_id", "circuit_display"]

    def get_circuit_display(self, obj) -> str | None:
        if not obj.circuit_id:
            return None
        adapter = self.context.get("circuit_adapter")
        return adapter.get_circuit_display(obj.circuit_id) if adapter else None

class ServiceOrderViewSet(NetBoxModelViewSet):
    queryset = ServiceOrder.objects.all()
    serializer_class = ServiceOrderSerializer

    def get_serializer_context(self):
        context = super().get_serializer_context()
        context["circuit_adapter"] = get_circuit_adapter()
        return context

    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        service = OrderProvisioningService.for_order_type(
            serializer.validated_data["order_type"],
            circuit_adapter=get_circuit_adapter(),
            location_adapter=get_location_adapter(),
        )
        order = service.create_order(serializer.validated_data)
        output = self.get_serializer(order)
        return Response(output.data, status=status.HTTP_201_CREATED)

For UI views, use NetBox generics (ObjectView, ObjectListView, ObjectEditView, ObjectDeleteView) — they handle permissions, templates, and navigation.

Form Patterns

• UI rendering only: No business logic in forms
• Conditional layout: Adapt fields based on create vs edit (self.instance.pk)
• FieldSet grouping: Organize sections with NetBox's FieldSet
• DynamicModelChoiceField: For cross-module lookups

from netbox.forms import NetBoxModelForm
from utilities.forms.fields import DynamicModelChoiceField

class ServiceOrderForm(NetBoxModelForm):
    fieldsets = (
        FieldSet("name", "order_type", name="Order"),
        FieldSet("circuit_id", name="References"),
    )

    class Meta:
        model = ServiceOrder
        fields = ["name", "order_type", "circuit_id"]

Background Jobs

from netbox.jobs import JobRunner

class ProvisioningSyncJob(JobRunner):
    """Background job with adapter injection."""

    class Meta:
        name = "Provisioning Sync"

    def run(self, data, commit):
        adapters = {
            "circuit_adapter": get_circuit_adapter(),
            "location_adapter": get_location_adapter(),
        }
        synced = ProvisioningSync.sync_all_pending(**adapters)
        self.log_info(f"Synced {synced} orders")

What Doesn't Belong in Models

Anti-Patterns

• Business logic in views or serializers — move to models or services
• Direct cross-plugin model imports — use adapters with import_string
• Mocking in tests — use in-memory adapters with reset()
• Models calling save() in business methods — let the caller persist
• Signals for business logic — use explicit service calls
• Mixing query and command in the same model method
• ForeignKey across plugin boundaries — use integer ID + adapter
• God services — keep services focused on one workflow each

Testing

Django-specific testing patterns (test clients, in-memory adapters, fixture design, factory patterns) are covered in the companion testing-django skill.