Quality Gates

# FORBIDDEN: sync ORM call inside async function
async def bad_view(request):
    user = User.objects.get(id=1)  # SynchronousOnlyOperation at runtime
 
# REQUIRED: use a-prefixed method
async def good_view(request):
    user = await User.objects.aget(id=1)

# FORBIDDEN: multiple writes without transaction
def bad_transfer(from_id: int, to_id: int, amount: Decimal):
    Account.objects.filter(id=from_id).update(balance=F("balance") - amount)
    Account.objects.filter(id=to_id).update(balance=F("balance") + amount)
    # If second update fails, data is inconsistent
 
# REQUIRED: wrap in transaction
from django.db import transaction
 
def good_transfer(from_id: int, to_id: int, amount: Decimal):
    with transaction.atomic():
        Account.objects.filter(id=from_id).update(balance=F("balance") - amount)
        Account.objects.filter(id=to_id).update(balance=F("balance") + amount)

# FORBIDDEN: N+1 query pattern
def bad_list_books():
    books = Book.objects.all()
    for book in books:
        print(book.author.name)  # N extra queries
 
# REQUIRED: eager loading
def good_list_books():
    books = Book.objects.select_related("author").all()
    for book in books:
        print(book.author.name)  # 1 query total

# FORBIDDEN: raw user input in extra/raw
def bad_search(query: str):
    return Book.objects.extra(where=[f"title LIKE '%{query}%'"])
 
# REQUIRED: parameterized queries
def good_search(query: str):
    return Book.objects.filter(title__icontains=query)

# Type checking (Foundation)
mypy . --strict --show-error-codes
 
# Code quality (Foundation + Production)
black . --check
isort . --check-only
flake8 . --max-complexity=10 --max-line-length=88
 
# Security (Production)
bandit -r src/ -f json
 
# Test coverage (Production)
pytest --cov=src --cov-fail-under=95 --maxfail=1

# Detect sync ORM in async functions (zero-tolerance)
# Look for .objects. calls without await/a-prefix in async def blocks
grep -rn "async def" src/ | while read line; do
  file=$(echo "$line" | cut -d: -f1)
  grep -n "\.objects\.\(get\|filter\|create\|update\|delete\|all\|count\|exists\)(" "$file" \
    | grep -v "await" | grep -v "\.a\(get\|filter\|create\|update\|delete\|all\|count\|exists\)("
done
 
# Detect missing transaction.atomic for bulk writes
grep -rn "\.update(" src/ | grep -v "transaction" | grep -v "test"
 
# Detect N+1 candidates: all() without select/prefetch
grep -rn "\.objects\.all()" src/ | grep -v "select_related\|prefetch_related"
 
# Pydantic v1 leftovers
grep -rn "\.parse_obj\|\.dict()\|@validator\b" src/

# .github/workflows/quality.yml
name: Quality Gates
on: [push, pull_request]
 
jobs:
  foundation:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -e .[dev]
      - run: mypy . --strict
      - run: black --check .
      - run: isort --check-only .
 
  production:
    needs: foundation
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env: { POSTGRES_DB: test, POSTGRES_PASSWORD: test }
        ports: ["5432:5432"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -e .[dev]
      - run: pytest --cov=src --cov-fail-under=95
      - run: bandit -r src/
      - run: flake8 . --max-complexity=10

import time
from prometheus_client import Counter, Histogram, Gauge
 
REQUEST_COUNT = Counter(
    "django_http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status"],
)
REQUEST_DURATION = Histogram(
    "django_http_request_duration_seconds",
    "HTTP request duration",
    ["method", "endpoint"],
)
ACTIVE_REQUESTS = Gauge(
    "django_http_active_requests",
    "Currently active HTTP requests",
)
 
 
class PrometheusMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
 
    def __call__(self, request):
        ACTIVE_REQUESTS.inc()
        start = time.perf_counter()
 
        response = self.get_response(request)
 
        duration = time.perf_counter() - start
        endpoint = self._get_endpoint(request)
 
        REQUEST_COUNT.labels(
            method=request.method,
            endpoint=endpoint,
            status=response.status_code,
        ).inc()
        REQUEST_DURATION.labels(
            method=request.method,
            endpoint=endpoint,
        ).observe(duration)
        ACTIVE_REQUESTS.dec()
 
        return response
 
    def _get_endpoint(self, request) -> str:
        if request.resolver_match:
            return request.resolver_match.route
        return "unknown"

class AsyncPrometheusMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
 
    async def __call__(self, request):
        ACTIVE_REQUESTS.inc()
        start = time.perf_counter()
 
        response = await self.get_response(request)
 
        duration = time.perf_counter() - start
        endpoint = self._get_endpoint(request)
 
        REQUEST_COUNT.labels(
            method=request.method,
            endpoint=endpoint,
            status=response.status_code,
        ).inc()
        REQUEST_DURATION.labels(
            method=request.method,
            endpoint=endpoint,
        ).observe(duration)
        ACTIVE_REQUESTS.dec()
 
        return response

MIDDLEWARE = [
    "project.middleware.PrometheusMiddleware",  # First — to capture all requests
    # ... other middleware
]