import threading
from contextlib import contextmanager
_state = threading.local()
def set_current_tenant(organization):
_state.tenant = organization
def get_current_tenant():
return getattr(_state, "tenant", None)
def clear_current_tenant():
if hasattr(_state, "tenant"):
del _state.tenant
@contextmanager
def tenant_scope(organization):
previous = get_current_tenant()
set_current_tenant(organization)
try:
yield
finally:
if previous is None:
clear_current_tenant()
else:
set_current_tenant(previous)
from django.http import HttpResponseForbidden
from .tenant_context import set_current_tenant, clear_current_tenant
class TenantMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
organization = self._resolve_organization(request)
if request.user.is_authenticated and organization is None:
return HttpResponseForbidden("No organization for this account.")
set_current_tenant(organization)
try:
return self.get_response(request)
finally:
clear_current_tenant()
def _resolve_organization(self, request):
user = getattr(request, "user", None)
if user is None or not user.is_authenticated:
return None
return getattr(user, "organization", None)
from django.db import models
from .tenant_context import get_current_tenant
class TenantManager(models.Manager):
def get_queryset(self):
queryset = super().get_queryset()
tenant = get_current_tenant()
if tenant is None:
return queryset.none()
return queryset.filter(organization=tenant)
class TenantModel(models.Model):
organization = models.ForeignKey(
"orgs.Organization",
on_delete=models.CASCADE,
related_name="%(class)s_set",
db_index=True,
)
objects = TenantManager()
all_objects = models.Manager()
class Meta:
abstract = True
def save(self, *args, **kwargs):
if self.organization_id is None:
tenant = get_current_tenant()
if tenant is None:
raise ValueError("Cannot save tenant object without an active tenant.")
self.organization = tenant
super().save(*args, **kwargs)
class Document(TenantModel):
title = models.CharField(max_length=255)
body = models.TextField(blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ("-created_at",)
def __str__(self):
return self.title
from django.contrib.auth.mixins import LoginRequiredMixin
from django.views.generic import ListView, CreateView
from django.urls import reverse_lazy
from .models import Document
class DocumentListView(LoginRequiredMixin, ListView):
model = Document # TenantManager already scopes this to the current org
template_name = "documents/list.html"
context_object_name = "documents"
class DocumentCreateView(LoginRequiredMixin, CreateView):
model = Document
fields = ("title", "body")
template_name = "documents/form.html"
success_url = reverse_lazy("documents:list")
Multi-tenant SaaS applications share one database across many customers, so the single most dangerous bug class is a query that leaks one tenant's rows to another. This snippet enforces isolation at the ORM layer so that every model query is automatically scoped to the current request's organization, rather than trusting each view to remember a .filter(org=...) call.
In tenant_context.py, a threading.local() object holds the active Organization for the duration of a request. The helper functions set_current_tenant, get_current_tenant, and clear_current_tenant are the only sanctioned way to touch that state. Thread-local storage is used because Django's synchronous request handling gives each worker thread one request at a time, so the tenant set at the start of a request stays correct until it is cleared. The tenant_scope context manager exists for background jobs and management commands, where there is no request to read from but code still needs an explicit tenant boundary.
In TenantMiddleware, the organization is resolved from the authenticated user once per request and stored via set_current_tenant. The try/finally around get_response guarantees clear_current_tenant runs even if the view raises, which prevents a leftover tenant from one request bleeding into the next when the worker thread is reused.
In models.py, TenantManager.get_queryset reads the current tenant and appends the organization filter transparently, so Document.objects.all() returns only the caller's rows. Keeping the unscoped Manager available as all_objects is important: admin tooling, cross-tenant reports, and the middleware's own lookups need an escape hatch, and hiding it entirely would push developers toward unsafe raw queries. TenantModel.save auto-stamps organization from the current tenant so callers cannot accidentally create orphaned or misattributed records.
The main trade-off is implicitness — a reader of a view cannot see the tenant filter, so the rule must be well documented and the all_objects manager audited. The pattern also relies on the middleware always running before any query; queries in middleware ordered before TenantMiddleware will see no tenant. When those constraints hold, isolation becomes a property of the framework instead of a checklist every engineer must remember.
Related snips
payload = {
sub: user.id,
iss: 'https://auth.example.com',
aud: 'codesnips-api',
exp: 15.minutes.from_now.to_i,
iat: Time.now.to_i,JWT issuance and verification without common footguns
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
from django.urls import path
from . import views
app_name = 'blog'
urlpatterns = [Django URL namespacing and reverse lookups
#!/usr/bin/env bash
set -euo pipefail
export VAULT_ADDR="https://vault.internal:8200"
export VAULT_TOKEN="${VAULT_TOKEN:?missing VAULT_TOKEN}"Secrets management with environment isolation and Vault
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";JWT access + refresh token rotation (conceptual)
import graphene
from graphene_django import DjangoObjectType
from blog.models import Post, Comment
class PostType(DjangoObjectType):Django GraphQL with Graphene
Share this code
Here's the card — post it anywhere.
