Skip to main content

C3 Component Diagram: Django Backend Components

Purpose: Detailed component-level architecture of the Django License Server backend, showing Django apps, models, viewsets, middleware, and service layers with multi-tenant architecture implementation.

Scope: Django application layer (backend container from C2)

Related Diagrams:

Mermaid Component Diagram​

Component Details​

1. Django Apps Structure​

apps.tenants​

Purpose: Core tenant management and multi-tenant infrastructure

Components:

  • models.py
    • Tenant - Organization model (name, slug, plan_tier, stripe_customer_id)
    • TenantModel - Abstract base class for tenant-scoped models
  • context.py
    • set_current_tenant() - Set tenant context for request
    • get_current_tenant() - Retrieve current tenant
    • clear_current_tenant() - Clear tenant context
    • tenant_context - Context manager for tenant-scoped operations
  • middleware.py
    • TenantMiddleware - Sets tenant from authenticated user
  • viewsets.py
    • TenantViewSet - Base viewset with automatic tenant filtering

Key Responsibilities:

  • Tenant context management (thread-safe via ContextVar)
  • PostgreSQL session variable management (set_current_tenant() SQL function)
  • Automatic tenant filtering for all queries
  • Middleware integration with authentication

apps.users​

Purpose: User management with tenant-scoped authentication

Components:

  • models.py
    • User(TenantModel, AbstractBaseUser) - Custom user model with tenant FK
    • UserRole - Enum: owner, admin, member, viewer
  • serializers.py
    • UserSerializer - User CRUD serialization
    • UserCreateSerializer - User registration with password hashing
  • viewsets.py
    • UserViewSet(TenantViewSet) - User management API
    • Role-based filtering (admins see all, members see self)
  • permissions.py
    • IsOwnerOrAdmin - Custom permission for user management

Key Responsibilities:

  • User authentication (email + password)
  • Tenant-scoped user management
  • Role-based access control (RBAC)
  • Permission validation per tenant

Example Code:

class User(TenantModel, AbstractBaseUser):
    tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)
    email = models.EmailField(unique=False)  # Unique per tenant, not global
    role = models.CharField(choices=UserRole.choices, default='member')

    class Meta:
        unique_together = [('tenant', 'email')]

apps.licenses​

Purpose: License session management and validation

Components:

  • models.py
    • LicenseSession(TenantModel) - Active license sessions
    • LicenseType - Enum: free, pro, team, enterprise
    • SessionStatus - Enum: active, expired, revoked
  • serializers.py
    • LicenseSessionSerializer - Session CRUD
    • LicenseValidationSerializer - Validation request/response
  • viewsets.py
    • LicenseSessionViewSet(TenantViewSet) - License API
    • @action methods: validate, heartbeat, release
  • services.py
    • LicenseService - License validation logic with Redis
    • SessionService - Session lifecycle management

Key Responsibilities:

  • License validation (<100ms p99)
  • Session heartbeat tracking (Redis TTL 6 min)
  • Atomic seat counting (Redis Lua scripts)
  • Cloud KMS license signing (RSA-4096)

Example Code:

class LicenseSession(TenantModel):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    session_token = models.CharField(max_length=255, unique=True)
    machine_id = models.CharField(max_length=255)
    license_type = models.CharField(choices=LicenseType.choices)
    expires_at = models.DateTimeField()

    def validate(self):
        """Validate session and update last_validated_at."""
        if not self.is_valid:
            return False
        self.last_validated_at = timezone.now()
        self.save(update_fields=['last_validated_at'])
        return True

apps.projects​

Purpose: Project management for license organization

Components:

  • models.py
    • Project(TenantModel) - User projects
    • ProjectStatus - Enum: active, archived
  • serializers.py
    • ProjectSerializer - Project CRUD
  • viewsets.py
    • ProjectViewSet(TenantViewSet) - Project API
    • Automatic owner assignment to current user

Key Responsibilities:

  • Project CRUD operations
  • Tenant quota enforcement (max_projects)
  • Project archiving/restoration

apps.audit​

Purpose: Comprehensive audit logging for compliance

Components:

  • models.py
    • AuditLog(TenantModel) - Audit trail entries
  • services.py
    • AuditService - Audit log creation with context
  • viewsets.py
    • AuditLogViewSet(TenantViewSet) - Read-only audit API

Key Responsibilities:

  • Automatic audit logging (create, update, delete, login)
  • Actor tracking (user_id, user_email, ip_address, user_agent)
  • Before/after change tracking (JSON diff)
  • Compliance reporting (SOC 2, GDPR, HIPAA)

Example Usage:

AuditLog.log(
    action='create',
    resource_type='project',
    resource_id=project.id,
    user=request.user,
    metadata={'project_name': project.name},
    request=request
)

apps.auth​

Purpose: Authentication and JWT token management

Components:

  • authentication.py
    • JWTAuthentication - Custom JWT auth backend
  • serializers.py
    • LoginSerializer - Email/password login
    • TokenSerializer - JWT token response
  • views.py
    • LoginView - JWT token generation
    • RefreshView - Token refresh
    • LogoutView - Token revocation

Key Responsibilities:

  • JWT token generation and validation
  • Integration with Identity Platform (OAuth2/OIDC)
  • Token refresh and revocation
  • Multi-provider auth (Google, GitHub)

2. Middleware Layer​

Execution Order (CRITICAL):

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',           # 1. HTTPS enforcement
    'corsheaders.middleware.CorsMiddleware',                   # 2. CORS headers
    'django.contrib.sessions.middleware.SessionMiddleware',    # 3. Session support
    'django.middleware.common.CommonMiddleware',               # 4. Common processing
    'django.middleware.csrf.CsrfViewMiddleware',               # 5. CSRF protection
    'django.contrib.auth.middleware.AuthenticationMiddleware', # 6. Set request.user
    'apps.tenants.middleware.TenantMiddleware',                # 7. Set request.tenant ⭐
    'django.contrib.messages.middleware.MessageMiddleware',    # 8. Messages
    'django.middleware.clickjacking.XFrameOptionsMiddleware',  # 9. Clickjacking protection
]

CRITICAL: TenantMiddleware MUST come AFTER AuthenticationMiddleware because it requires request.user to be set.

TenantMiddleware Workflow​

class TenantMiddleware(MiddlewareMixin):
    def process_request(self, request):
        clear_current_tenant()  # Clear previous context

        if request.user and request.user.is_authenticated:
            if hasattr(request.user, 'tenant'):
                set_current_tenant(request.user.tenant)  # Set Python + PostgreSQL context
                request.tenant = request.user.tenant

    def process_response(self, request, response):
        clear_current_tenant()  # Always clear after request
        return response

Side Effects:

  • Sets Python ContextVar[Tenant] for application code
  • Sets PostgreSQL session variable app.current_tenant_id for RLS enforcement
  • Attaches request.tenant for easy access in views

3. Tenant Context Management (Thread-Safe)​

Implementation:

# apps/tenants/context.py

from contextvars import ContextVar
from typing import Optional

_current_tenant: ContextVar[Optional[Tenant]] = ContextVar('current_tenant', default=None)

def set_current_tenant(tenant: Optional[Tenant]) -> None:
    """Set current tenant for this context (thread-safe)."""
    _current_tenant.set(tenant)

    # Also set PostgreSQL session variable for RLS
    if tenant:
        from django.db import connection
        with connection.cursor() as cursor:
            cursor.execute("SELECT set_current_tenant(%s)", [str(tenant.id)])

def get_current_tenant() -> Optional[Tenant]:
    """Get current tenant from context."""
    return _current_tenant.get()

def clear_current_tenant() -> None:
    """Clear current tenant context."""
    _current_tenant.set(None)

    from django.db import connection
    with connection.cursor() as cursor:
        cursor.execute("SELECT clear_current_tenant()")

Context Manager for Async Tasks:

class tenant_context:
    """Context manager for temporarily setting tenant context."""

    def __init__(self, tenant: Optional[Tenant]):
        self.tenant = tenant
        self.previous_tenant = None

    def __enter__(self):
        self.previous_tenant = get_current_tenant()
        set_current_tenant(self.tenant)
        return self.tenant

    def __exit__(self, exc_type, exc_val, exc_tb):
        set_current_tenant(self.previous_tenant)

# Usage in Celery tasks
@shared_task
def send_license_expiry_email(license_id):
    license = LicenseSession.objects.get(id=license_id)

    with tenant_context(license.tenant):
        # All queries here are tenant-scoped
        user = license.user
        EmailService.send_expiry_notification(user.email)

4. Model Layer - django-multitenant Integration​

TenantModel Base Class​

# apps/tenants/models.py

from django_multitenant.models import TenantModel as BaseTenantModel

class TenantModel(BaseTenantModel):
    """
    Base class for all tenant-scoped models.

    Features:
    - Automatic tenant FK enforcement
    - Auto-filtering by current tenant
    - Save validation (requires tenant context)
    """

    tenant = models.ForeignKey(
        Tenant,
        on_delete=models.CASCADE,
        related_name='%(class)ss',  # e.g., tenant.users, tenant.projects
        db_index=True
    )

    class Meta:
        abstract = True

    def save(self, *args, **kwargs):
        # Auto-set tenant from context if not provided
        if not self.tenant_id:
            from .context import get_current_tenant
            current_tenant = get_current_tenant()
            if current_tenant:
                self.tenant_id = current_tenant.id
            else:
                raise ValueError(
                    f"Cannot save {self.__class__.__name__} without tenant context. "
                    "Set tenant explicitly or use TenantMiddleware."
                )
        super().save(*args, **kwargs)

Automatic Filtering:

# When tenant context is set, ALL queries auto-filter

set_current_tenant(my_tenant)

# These queries automatically filter to my_tenant:
User.objects.all()  # Returns only my_tenant's users
Project.objects.filter(status='active')  # Returns only my_tenant's active projects
LicenseSession.objects.get(id=some_id)  # Only returns if session belongs to my_tenant

# Trying to access another tenant's data returns empty queryset or DoesNotExist

PostgreSQL Row-Level Security Enforcement:

Even if application code has bugs, PostgreSQL RLS prevents cross-tenant access:

-- RLS Policy on users table
CREATE POLICY users_tenant_isolation ON users
    USING (tenant_id = current_setting('app.current_tenant_id')::UUID);

-- Result: Raw SQL queries ALSO respect tenant isolation
SELECT * FROM users;  -- Only returns current tenant's users
UPDATE users SET username = 'hacked' WHERE id = 'other-tenant-user-id';  -- Silently fails (0 rows affected)

5. ViewSet Layer - Django REST Framework​

TenantViewSet Base Class​

# apps/api/viewsets.py

from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated
from apps.tenants.context import get_current_tenant

class TenantViewSet(viewsets.ModelViewSet):
    """
    Base viewset for tenant-scoped resources.

    Features:
    - Automatic tenant filtering
    - Automatic tenant assignment on create
    - Permission enforcement
    """

    permission_classes = [IsAuthenticated]

    def get_queryset(self):
        """Filter queryset to current tenant."""
        queryset = super().get_queryset()

        tenant = get_current_tenant()
        if tenant:
            # Explicit filter for clarity (also enforced by RLS)
            queryset = queryset.filter(tenant=tenant)

        return queryset

    def perform_create(self, serializer):
        """Set tenant on create."""
        tenant = get_current_tenant()
        if not tenant:
            from rest_framework.exceptions import PermissionDenied
            raise PermissionDenied("No tenant context available")

        serializer.save(tenant=tenant)

Example ViewSets​

UserViewSet:

class UserViewSet(TenantViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

    def get_queryset(self):
        queryset = super().get_queryset()
        user = self.request.user

        # Role-based filtering
        if user.is_owner or user.is_admin:
            return queryset  # See all users in tenant
        else:
            return queryset.filter(id=user.id)  # See only self

LicenseSessionViewSet:

from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework import status

class LicenseSessionViewSet(TenantViewSet):
    queryset = LicenseSession.objects.all()
    serializer_class = LicenseSessionSerializer

    @action(detail=False, methods=['post'])
    def validate(self, request):
        """Validate license and acquire seat."""
        serializer = LicenseValidationSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        # Use service layer for business logic
        result = LicenseService.validate_license(
            user=request.user,
            machine_id=serializer.validated_data['machine_id'],
            tenant=get_current_tenant()
        )

        return Response(result, status=status.HTTP_200_OK)

    @action(detail=True, methods=['post'])
    def heartbeat(self, request, pk=None):
        """Update session heartbeat."""
        session = self.get_object()
        session.validate()

        return Response({'status': 'ok', 'expires_at': session.expires_at})

    @action(detail=True, methods=['post'])
    def release(self, request, pk=None):
        """Release license seat."""
        session = self.get_object()
        SessionService.release_seat(session)

        return Response({'status': 'released'})

6. Service Layer​

Purpose: Encapsulate business logic, external integrations, and complex operations.

LicenseService​

# apps/licenses/services.py

from django.utils import timezone
from apps.tenants.context import get_current_tenant
import redis
import uuid

class LicenseService:
    """License validation and seat management."""

    @classmethod
    def validate_license(cls, user, machine_id, tenant):
        """
        Validate license and acquire seat.

        Steps:
        1. Check tenant license quota
        2. Atomic seat acquisition in Redis
        3. Create session in PostgreSQL
        4. Sign license with Cloud KMS

        Returns:
            dict: {
                'valid': bool,
                'session_token': str,
                'signed_license': str,
                'expires_at': datetime,
                'features': list
            }
        """
        from apps.licenses.models import LicenseSession

        # 1. Check tenant quota
        active_sessions = LicenseSession.objects.filter(
            tenant=tenant,
            status='active'
        ).count()

        if active_sessions >= tenant.max_users:
            return {'valid': False, 'error': 'No available seats'}

        # 2. Atomic seat acquisition in Redis
        redis_client = get_redis_client()
        seat_key = f"tenant:{tenant.id}:seats"

        # Lua script for atomic increment with limit check
        lua_script = """
        local current = tonumber(redis.call('GET', KEYS[1]) or 0)
        local max = tonumber(ARGV[1])
        if current < max then
            redis.call('INCR', KEYS[1])
            return 1
        else
            return 0
        end
        """

        acquired = redis_client.eval(lua_script, 1, seat_key, tenant.max_users)

        if not acquired:
            return {'valid': False, 'error': 'No available seats (race condition)'}

        # 3. Create session
        session_token = str(uuid.uuid4())
        expires_at = timezone.now() + timezone.timedelta(hours=8)

        session = LicenseSession.objects.create(
            tenant=tenant,
            user=user,
            session_token=session_token,
            machine_id=machine_id,
            license_type=tenant.plan_tier,
            features=cls._get_features_for_tier(tenant.plan_tier),
            expires_at=expires_at
        )

        # 4. Sign license with Cloud KMS
        signed_license = cls._sign_license(session)

        # 5. Set Redis TTL for automatic cleanup (6 min)
        session_key = f"session:{session.id}"
        redis_client.setex(session_key, 360, session_token)

        return {
            'valid': True,
            'session_token': session_token,
            'signed_license': signed_license,
            'expires_at': expires_at,
            'features': session.features
        }

    @classmethod
    def _sign_license(cls, session):
        """Sign license with Cloud KMS RSA-4096."""
        from google.cloud import kms

        client = kms.KeyManagementServiceClient()
        key_name = f"projects/{PROJECT_ID}/locations/global/keyRings/licenses/cryptoKeys/signing/cryptoKeyVersions/1"

        # Create license payload
        payload = {
            'session_id': str(session.id),
            'user_id': str(session.user_id),
            'tenant_id': str(session.tenant_id),
            'machine_id': session.machine_id,
            'expires_at': session.expires_at.isoformat(),
            'features': session.features
        }

        import json
        import base64

        message = json.dumps(payload).encode('utf-8')

        # Sign with KMS
        response = client.asymmetric_sign(
            request={'name': key_name, 'digest': {'sha256': hashlib.sha256(message).digest()}}
        )

        signature = base64.b64encode(response.signature).decode('utf-8')

        return {
            'payload': payload,
            'signature': signature,
            'algorithm': 'RSA_SIGN_PKCS1_4096_SHA256'
        }

SessionService​

class SessionService:
    """Session lifecycle management."""

    @classmethod
    def release_seat(cls, session):
        """Release license seat and decrement Redis counter."""
        from apps.licenses.models import SessionStatus

        # 1. Update session status
        session.status = SessionStatus.REVOKED
        session.revoked_at = timezone.now()
        session.save()

        # 2. Decrement Redis seat counter
        redis_client = get_redis_client()
        seat_key = f"tenant:{session.tenant_id}:seats"
        redis_client.decr(seat_key)

        # 3. Delete Redis session key
        session_key = f"session:{session.id}"
        redis_client.delete(session_key)

        # 4. Audit log
        AuditLog.log(
            action='release',
            resource_type='license_session',
            resource_id=session.id,
            user=session.user,
            metadata={'machine_id': session.machine_id}
        )

7. Background Tasks - Celery​

Celery Configuration:

# config/celery.py

from celery import Celery
import os

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.production')

app = Celery('license_server')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# Redis as broker and result backend
app.conf.broker_url = 'redis://redis:6379/0'
app.conf.result_backend = 'redis://redis:6379/0'

Periodic Tasks:

# apps/licenses/tasks.py

from celery import shared_task
from apps.tenants.context import tenant_context
from apps.tenants.models import Tenant
from apps.licenses.models import LicenseSession, SessionStatus
from django.utils import timezone

@shared_task
def session_cleanup_task():
    """Expire old sessions (runs every 5 minutes)."""

    # Process all tenants
    for tenant in Tenant.objects.filter(status='active'):
        with tenant_context(tenant):
            # Find expired sessions
            expired_sessions = LicenseSession.objects.filter(
                status=SessionStatus.ACTIVE,
                expires_at__lt=timezone.now()
            )

            for session in expired_sessions:
                session.status = SessionStatus.EXPIRED
                session.save()

                # Release seat in Redis
                SessionService.release_seat(session)

@shared_task
def audit_archive_task():
    """Archive audit logs older than 90 days (runs daily)."""
    from apps.audit.models import AuditLog

    cutoff = timezone.now() - timezone.timedelta(days=90)

    for tenant in Tenant.objects.all():
        with tenant_context(tenant):
            old_logs = AuditLog.objects.filter(created_at__lt=cutoff)

            # Archive to Cloud Storage
            # ... (implementation details)

            old_logs.delete()

@shared_task
def metrics_update_task():
    """Update Prometheus metrics (runs every 1 minute)."""
    from apps.tenants.metrics import update_tenant_metrics

    for tenant in Tenant.objects.filter(status='active'):
        update_tenant_metrics(tenant)

Celery Beat Schedule:

# config/settings/base.py

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'session-cleanup': {
        'task': 'apps.licenses.tasks.session_cleanup_task',
        'schedule': 300.0,  # Every 5 minutes
    },
    'audit-archive': {
        'task': 'apps.licenses.tasks.audit_archive_task',
        'schedule': crontab(hour=2, minute=0),  # 2 AM daily
    },
    'metrics-update': {
        'task': 'apps.licenses.tasks.metrics_update_task',
        'schedule': 60.0,  # Every 1 minute
    },
}

8. Utilities​

validators.py​

# apps/common/validators.py

from django.core.exceptions import ValidationError
import re

def validate_machine_id(value):
    """Validate machine ID format (UUID or SHA-256 hash)."""
    uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'
    sha256_pattern = r'^[0-9a-f]{64}$'

    if not (re.match(uuid_pattern, value) or re.match(sha256_pattern, value)):
        raise ValidationError('Invalid machine_id format')

def validate_tenant_slug(value):
    """Validate tenant slug (lowercase alphanumeric + hyphens)."""
    if not re.match(r'^[a-z0-9-]+$', value):
        raise ValidationError('Slug must be lowercase alphanumeric with hyphens')

permissions.py​

# apps/common/permissions.py

from rest_framework import permissions

class IsOwnerOrAdmin(permissions.BasePermission):
    """Allow access to tenant owners and admins."""

    def has_permission(self, request, view):
        return request.user and request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        if not request.user.is_authenticated:
            return False

        # Owners and admins have full access
        if request.user.is_owner or request.user.is_admin:
            return True

        # Members can only access their own objects
        if hasattr(obj, 'user'):
            return obj.user == request.user

        if hasattr(obj, 'owner'):
            return obj.owner == request.user

        return False

class IsTenantMember(permissions.BasePermission):
    """Allow access to any tenant member."""

    def has_permission(self, request, view):
        return request.user and request.user.is_authenticated and hasattr(request.user, 'tenant')

exceptions.py​

# apps/common/exceptions.py

from rest_framework.exceptions import APIException
from rest_framework import status

class NoSeatsAvailable(APIException):
    status_code = status.HTTP_403_FORBIDDEN
    default_detail = 'No license seats available'
    default_code = 'no_seats_available'

class TenantQuotaExceeded(APIException):
    status_code = status.HTTP_403_FORBIDDEN
    default_detail = 'Tenant quota exceeded'
    default_code = 'quota_exceeded'

class InvalidLicenseSignature(APIException):
    status_code = status.HTTP_401_UNAUTHORIZED
    default_detail = 'Invalid license signature'
    default_code = 'invalid_signature'

Data Flow Examples​

License Validation Flow​

1. Client → POST /api/v1/licenses/validate
   {
       "machine_id": "abc123...",
       "hardware_fingerprint": {...}
   }

2. Middleware → TenantMiddleware
   - AuthenticationMiddleware sets request.user
   - TenantMiddleware sets request.tenant from request.user.tenant
   - set_current_tenant(tenant) → Python ContextVar + PostgreSQL session var

3. ViewSet → LicenseSessionViewSet.validate()
   - Permission check: IsAuthenticated
   - Serializer validation
   - Call LicenseService.validate_license()

4. Service → LicenseService.validate_license()
   - Check PostgreSQL quota (tenant.max_users)
   - Atomic seat acquisition in Redis (Lua script)
   - Create LicenseSession in PostgreSQL (auto-scoped to tenant)
   - Sign license with Cloud KMS (RSA-4096)

5. Response → Client
   {
       "valid": true,
       "session_token": "uuid...",
       "signed_license": {
           "payload": {...},
           "signature": "base64...",
           "algorithm": "RSA_SIGN_PKCS1_4096_SHA256"
       },
       "expires_at": "2025-12-01T12:00:00Z",
       "features": ["feature1", "feature2"]
   }

6. Middleware → TenantMiddleware.process_response()
   - clear_current_tenant() → Python + PostgreSQL

Celery Background Task Flow​

1. Celery Beat → Triggers session_cleanup_task every 5 minutes

2. Task Execution → session_cleanup_task()
   - Iterate all active tenants

3. Per Tenant → with tenant_context(tenant):
   - set_current_tenant(tenant)
   - Query expired sessions (Django ORM auto-filters by tenant)
   - Update session status to EXPIRED
   - Release seat in Redis (decrement counter)
   - clear_current_tenant()

4. Result → All tenants processed independently

Technology Integration​

Django REST Framework Configuration​

# config/settings/base.py

REST_FRAMEWORK = {
    # Authentication
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.SessionAuthentication',
        'apps.auth.authentication.JWTAuthentication',
    ],

    # Permissions
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],

    # Pagination
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.LimitOffsetPagination',
    'PAGE_SIZE': 50,
    'MAX_PAGE_SIZE': 1000,

    # Filtering
    'DEFAULT_FILTER_BACKENDS': [
        'django_filters.rest_framework.DjangoFilterBackend',
        'rest_framework.filters.SearchFilter',
        'rest_framework.filters.OrderingFilter',
    ],

    # Throttling
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/hour',
        'user': '10000/hour',
    },
}

django-multitenant Configuration​

# config/settings/base.py

INSTALLED_APPS = [
    # ...
    'django_multitenant',
    # ...
]

# Tenant FK field name (default: 'tenant')
MULTITENANT_RELATIVE_NAME = 'tenant'

Testing Strategy​

Unit Tests​

# apps/tenants/tests/test_context.py

def test_tenant_context_isolation():
    """Verify tenant context is isolated per request."""
    tenant1 = Tenant.objects.create(name="Tenant 1")
    tenant2 = Tenant.objects.create(name="Tenant 2")

    set_current_tenant(tenant1)
    assert get_current_tenant() == tenant1

    set_current_tenant(tenant2)
    assert get_current_tenant() == tenant2

    clear_current_tenant()
    assert get_current_tenant() is None

Integration Tests​

# apps/api/tests/test_license_api.py

def test_license_validation_endpoint():
    """Test license validation API with tenant isolation."""
    client = APIClient()
    client.force_authenticate(user=self.user1)

    response = client.post('/api/v1/licenses/validate', {
        'machine_id': 'test-machine-123'
    })

    assert response.status_code == 200
    assert response.data['valid'] is True
    assert 'signed_license' in response.data

Performance Tests​

# apps/tenants/tests/test_performance.py

def test_query_performance_with_rls():
    """Measure query performance with RLS enabled."""
    timings = []

    for tenant in self.tenants[:10]:
        set_current_tenant(tenant)
        start = time.perf_counter()

        sessions = LicenseSession.objects.filter(status='active').select_related('user').all()
        count = len(sessions)

        elapsed = (time.perf_counter() - start) * 1000
        timings.append(elapsed)

    avg_time = statistics.mean(timings)
    assert avg_time < 50, "Average query time should be <50ms"

Deployment Considerations​

Environment Variables​

# Django settings
DJANGO_SETTINGS_MODULE=config.settings.production
SECRET_KEY=<secure-random-key>

# Database (Cloud SQL)
DB_HOST=/cloudsql/coditect-cloud-infra:us-central1:licenses-db
DB_PORT=5432
DB_NAME=licenses
DB_USER=app_user
DB_PASSWORD=<from-secret-manager>

# Redis (Memorystore)
REDIS_HOST=10.0.0.3
REDIS_PORT=6379

# Celery
CELERY_BROKER_URL=redis://10.0.0.3:6379/0
CELERY_RESULT_BACKEND=redis://10.0.0.3:6379/0

# GCP
GCP_PROJECT_ID=coditect-cloud-infra
GCP_REGION=us-central1
KMS_KEY_RING=licenses
KMS_SIGNING_KEY=signing

# CORS
CORS_ALLOWED_ORIGINS=https://app.coditect.ai,https://admin.coditect.ai

Kubernetes Deployment​

# kubernetes/django-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: django-backend
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: django
        image: gcr.io/coditect-cloud-infra/license-server:latest
        command: ["gunicorn", "config.wsgi:application", "--bind", "0.0.0.0:8000", "--workers", "4"]
        env:
        - name: DJANGO_SETTINGS_MODULE
          value: config.settings.production
        - name: DB_PASSWORD
          valueFrom:
            secretKeyRef:
              name: db-credentials
              key: password

Monitoring & Observability​

Prometheus Metrics​

# Exposed at /metrics endpoint

# Request metrics per tenant
tenant_requests_total{tenant_id="...", method="POST", path="/api/v1/licenses/validate", status="200"} 1234
tenant_request_duration_milliseconds{tenant_id="...", method="POST", path="/api/v1/licenses/validate"} 45.2

# License metrics per tenant
tenant_active_users{tenant_id="..."} 87
tenant_license_sessions{tenant_id="...", status="active"} 45

Logging​

# Structured JSON logging

{
    "timestamp": "2025-11-30T12:34:56Z",
    "level": "INFO",
    "logger": "apps.licenses.services",
    "message": "License validated successfully",
    "tenant_id": "uuid...",
    "user_id": "uuid...",
    "machine_id": "abc123...",
    "session_id": "uuid...",
    "duration_ms": 42.5
}

Last Updated: 2025-11-30 Diagram Type: C3 Component (Mermaid) Scope: Django Backend - Application Components