Skip to content

Global Project Reference Patterns

IMPORTANT: Check this file FIRST before implementing ANY feature. These patterns apply to ALL components.

Table of Contents

  1. Code Organization
  2. Error Handling
  3. Logging and Monitoring
  4. Database Patterns
  5. API Design
  6. Security
  7. Testing
  8. Documentation
  9. Environment Configuration
  10. Anti-Patterns

Code Organization

Directory Structure

Pattern: 

[backend/frontend]/
├── [framework-specific structure]
├── models/          # Data models
├── services/        # Business logic
├── controllers/     # Request handlers (backend) or components (frontend)
├── utils/           # Shared utilities
├── config/          # Configuration
└── tests/           # Test files mirroring source structure

Guidelines: - Mirror test directory structure to source - One component/module per file - Group related functionality in subdirectories - Keep files under 500 lines (split if larger)

Naming Conventions

Files: - kebab-case.ext for all files - Test files: [name].test.ext or test_[name].ext

Code: - Functions/methods: camelCase (JavaScript/TypeScript) or snake_case (Python) - Classes: PascalCase - Constants: UPPER_SNAKE_CASE - Private/internal: prefix with _ (Python) or use # (JavaScript private fields)

Import Organization

Pattern: 

# Python example
# 1. Standard library
import os
import sys

# 2. Third-party libraries
from fastapi import FastAPI
from sqlalchemy import Column

# 3. Local imports
from .models import User
from .services import UserService


// TypeScript example
// 1. React/framework imports
import { useState, useEffect } from 'react';

// 2. Third-party libraries
import axios from 'axios';

// 3. Local imports
import { UserService } from '@/services/user';
import type { User } from '@/types';

Error Handling

Pattern: Structured Error Responses

Backend: 

# Standard error response format
{
    "error": {
        "code": "VALIDATION_ERROR",
        "message": "User-friendly error message",
        "details": {
            "field": "email",
            "issue": "Invalid email format"
        },
        "timestamp": "2025-01-15T12:34:56Z",
        "request_id": "uuid-here"
    }
}

Error Categories: - VALIDATION_ERROR - Input validation failures - AUTHENTICATION_ERROR - Auth failures - AUTHORIZATION_ERROR - Permission denied - NOT_FOUND - Resource not found - CONFLICT - State conflict (duplicate, version mismatch) - INTERNAL_ERROR - Server errors

Pattern: Error Logging

Always log: - Error type and message - Request ID for tracing - User ID (if authenticated) - Stack trace (server-side only, never to client) - Context (what operation was being attempted)

Example (using structlog): 

logger.error(
    "Failed to create user",
    error_type="ValidationError",
    user_input=sanitized_input,
    request_id=request_id,
    src_ip=request.client.host,
    exc_info=True  # Include stack trace
)

Anti-Pattern: Generic Error Messages

❌ Don't: 

raise Exception("Something went wrong")
return {"error": "Error occurred"}

✅ Do: 

raise ValidationError("Email format invalid: missing @ symbol")
return {
    "error": {
        "code": "VALIDATION_ERROR",
        "message": "Email address is invalid",
        "details": {"field": "email", "issue": "missing @ symbol"}
    }
}

Logging and Monitoring

See: docs/prp/logging-prp.md for comprehensive logging implementation patterns, code examples, and anti-patterns.

Log Levels

  • DEBUG - Detailed development info (never in production)
  • INFO - Routine operations (user login, record created)
  • WARNING - Unexpected but handled (rate limit approached, deprecated feature used)
  • ERROR - Operation failed but app continues
  • CRITICAL - System failure, immediate attention required

Key Requirements

  • Single-line JSON format with timestamp first
  • Use structlog with keyword arguments (not extra={})
  • Always include: request_id, user_id, src_ip, status_code, duration_ms
  • Use unbind_contextvars() in async code (not clear_contextvars())
  • Never log passwords, tokens, or API keys

Database Patterns

Pattern: Migration Management

Guidelines: - Never modify existing migrations (create new one to fix) - Use descriptive migration names: add_user_email_verification_column - Include both upgrade and downgrade paths - Test migrations on copy of production data before deploying

Verification: 

# ALWAYS check live database schema
./scripts/db-schema.sh [table_name]

# Compare with models before writing code

Pattern: Query Optimization

Guidelines: - Use indexes for frequently queried columns - Avoid N+1 queries (use joins or eager loading) - Paginate large result sets - Use database-level constraints (foreign keys, unique, not null)

Example: 

# ❌ N+1 query problem
users = session.query(User).all()
for user in users:
    posts = session.query(Post).filter_by(user_id=user.id).all()  # N queries

# ✅ Eager loading
users = session.query(User).options(joinedload(User.posts)).all()  # 1 query

Pattern: Transactions

Use transactions for: - Multi-step operations that must complete together - Operations that modify multiple tables - Any operation where partial completion would leave invalid state

Example: 

with session.begin():
    user = User(email=email)
    session.add(user)

    profile = Profile(user_id=user.id, name=name)
    session.add(profile)

    # Both commit together or both rollback

API Design

Pattern: RESTful Endpoints

Standard CRUD: - GET /resource - List resources (paginated) - GET /resource/{id} - Get single resource - POST /resource - Create new resource - PUT /resource/{id} - Update entire resource - PATCH /resource/{id} - Partial update - DELETE /resource/{id} - Delete resource

Pattern: Response Format

Success: 

{
    "data": { /* resource or array */ },
    "meta": {
        "page": 1,
        "per_page": 20,
        "total": 100
    }
}

Error: (See Error Handling section)

Pattern: Versioning

Use URL versioning: 

/api/v1/users
/api/v2/users

When to create new version: - Breaking changes to request/response format - Removing fields - Changing field types

Maintain compatibility: - Adding optional fields is NOT a breaking change - Deprecate old versions with 6-month notice

Security

Pattern: Authentication

Guidelines: - Use secure, httpOnly cookies for session tokens (web) - Use JWT with short expiration for API tokens - Implement refresh token rotation - Hash passwords with bcrypt/argon2 (never MD5/SHA)

Pattern: Authorization

Check permissions at multiple levels: 1. API endpoint (route guard) 2. Service layer (business logic) 3. Database (row-level security if supported)

Example: 

@require_auth
async def update_user(user_id: int, current_user: User):
    # Check ownership
    if user_id != current_user.id and not current_user.is_admin:
        raise AuthorizationError("Cannot modify other users")

    # Proceed with update

Pattern: Input Validation

ALWAYS validate: - Data types - Required fields - Format (email, phone, etc.) - Length limits - Allowed values (enums)

Validate at: 1. Client-side - Better UX 2. Server-side - Security (NEVER trust client)

Anti-Pattern: SQL Injection

❌ Never: 

query = f"SELECT * FROM users WHERE email = '{email}'"

✅ Always use parameterized queries: 

query = "SELECT * FROM users WHERE email = :email"
session.execute(query, {"email": email})

Testing

Test Coverage Requirements

  • Critical paths: 100% (authentication, payment, data integrity)
  • Business logic: 90%+
  • Utilities: 80%+
  • UI components: 70%+

Pattern: Test Organization

Mirror source structure: 

src/services/user.py
tests/services/test_user.py

Pattern: Test Naming

Format: test_[function]_[scenario]_[expected]

Examples: 

def test_create_user_valid_input_returns_user():
    ...

def test_create_user_duplicate_email_raises_error():
    ...

def test_login_invalid_password_returns_401():
    ...

Pattern: Test Data

Use fixtures for reusable test data: 

@pytest.fixture
def sample_user():
    return User(email="test@example.com", name="Test User")

def test_user_creation(sample_user):
    assert sample_user.email == "test@example.com"

CRITICAL: See TEST_CONFIG.md before writing tests.

Documentation

Code Comments

When to comment: - Complex algorithms or business logic - Non-obvious workarounds - Security considerations - Performance optimizations

When NOT to comment: - Obvious code ("increment counter") - Restating what code does - Outdated comments (update or delete)

Docstrings

Required for: - All public functions/methods - All classes - All modules

Format: 

def calculate_total(items: List[Item], tax_rate: float) -> Decimal:
    """
    Calculate total price including tax.

    Args:
        items: List of items to total
        tax_rate: Tax rate as decimal (0.08 for 8%)

    Returns:
        Total price including tax

    Raises:
        ValueError: If tax_rate is negative
    """

Environment Configuration

Pattern: Environment Variables

Structure: 

# .env file (never commit!)
DATABASE_URL=postgresql://...
API_KEY=secret_key_here
DEBUG=false

Access: 

import os
from dotenv import load_dotenv

load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")

Pattern: Configuration Validation

Validate at startup: 

required_vars = ["DATABASE_URL", "API_KEY", "SECRET_KEY"]
missing = [var for var in required_vars if not os.getenv(var)]
if missing:
    raise EnvironmentError(f"Missing required env vars: {missing}")

Anti-Patterns

❌ Magic Numbers

Don't: 

if user.age > 18:
    ...

Do: 

MINIMUM_AGE = 18
if user.age > MINIMUM_AGE:
    ...

❌ God Objects/Functions

Don't: - 500-line functions - Classes with 50+ methods - Functions doing 10 different things

Do: - Split into smaller, focused functions - Follow Single Responsibility Principle - Extract related functionality to separate classes

❌ Premature Optimization

Don't: - Optimize before measuring - Add complexity for hypothetical performance gains - Cache everything "just in case"

Do: - Profile to find actual bottlenecks - Optimize critical paths only - Measure impact of optimization

❌ Ignoring Errors

Don't: 

try:
    risky_operation()
except:
    pass  # Silent failure

Do: 

try:
    risky_operation(entity_id=entity_id)
except SpecificException as e:
    logger.error(
        "Operation failed",
        error_type=type(e).__name__,
        error_message=str(e),
        entity_id=entity_id,
        operation="risky_operation",
        exc_info=True
    )
    raise  # Re-raise or handle appropriately

Summary Checklist

Before implementing ANY feature, verify:

  • [ ] Checked global.md (this file)
  • [ ] Checked component-specific PRP (if exists)
  • [ ] Following code organization patterns
  • [ ] Error handling implemented
  • [ ] Logging configured
  • [ ] Database patterns followed
  • [ ] API design consistent
  • [ ] Security considerations addressed
  • [ ] Tests planned/written
  • [ ] Documentation complete
  • [ ] Environment variables configured

Project-Specific Patterns

Edge-to-Central Authentication

Pattern: API Key Authentication

All edge collector communication with the central server uses API keys over TLS.

HTTP Requests (all edge-to-central communication): 

# Edge collector - common headers for all requests
headers = {"X-API-Key": config.api_key}

# Detection upload (batch)
response = await client.post(
    f"{central_url}/api/v1/detections/batch",
    headers=headers,
    json={"detections": detections}
)

# Heartbeat (also polls for pending commands)
response = await client.post(
    f"{central_url}/api/v1/heartbeat",
    headers=headers,
    json={"uptime_sec": uptime, "queue_depth": queue.size(), ...}
)
pending_commands = response.json().get("pending_commands", [])

# Command acknowledgment
response = await client.post(
    f"{central_url}/api/v1/commands/{cmd_id}/ack",
    headers=headers,
    json={"status": "success", "message": "Config applied"}
)

Central Server Validation: 

async def verify_api_key(api_key: str = Header(alias="X-API-Key")) -> Collector:
    collector = await db.get_collector_by_api_key(api_key)
    if not collector or not collector.is_active:
        raise HTTPException(401, "Invalid or inactive API key")
    return collector

@app.post("/api/v1/detections/batch")
async def receive_detections(
    detections: List[Detection],
    collector: Collector = Depends(verify_api_key)
):
    # collector is authenticated and identified
    ...

Key Generation: 

import secrets

def generate_api_key() -> str:
    """Generate a secure API key for a collector."""
    return f"alpr_{secrets.token_urlsafe(32)}"

Detection Schema

Pattern: Normalized Detection Format

All camera adapters normalize detections to this standard schema before buffering:

@dataclass
class Detection:
    id: str                 # UUID v4
    collector_id: str       # Edge collector identifier
    timestamp: datetime     # UTC, ISO 8601
    camera: CameraInfo
    plate: PlateInfo
    vehicle: VehicleInfo
    images: ImageSet
    raw: dict               # Original vendor payload (optional)

@dataclass
class CameraInfo:
    id: str                 # Local camera identifier
    mac: str                # Normalized MAC (lowercase, no separators)

@dataclass
class PlateInfo:
    text: str               # Plate number (uppercase, no spaces)
    confidence: float       # 0.0 - 1.0
    country: str | None     # ISO country code
    state: str | None       # State/region code

@dataclass
class VehicleInfo:
    direction: str | None = None      # "entering" | "exiting" | "unknown"
    vehicle_type: str | None = None   # "car" | "truck" | "motorcycle" | "bus" | "van"
    color: str | None = None
    brand: str | None = None          # Vehicle make (e.g., "Toyota", "Ford")
    model: str | None = None          # Vehicle model (e.g., "Camry", "F-150")

@dataclass
class ImageSet:
    full: bytes | None      # Full scene image (JPEG)
    plate_crop: bytes | None  # Plate close-up (JPEG)

MAC Address Normalization: 

def normalize_mac(mac: str) -> str:
    """Normalize MAC address to lowercase, no separators."""
    return mac.lower().replace(":", "").replace("-", "")

Plate Number Normalization: 

def normalize_plate(plate: str) -> str:
    """Normalize plate number to uppercase, no spaces."""
    return plate.upper().replace(" ", "").replace("-", "")

Heartbeat & Command Polling Patterns

Edge sends heartbeat every 60 seconds. Central responds with any pending commands.

Heartbeat Request (Edge → Central): 

POST /api/v1/heartbeat
{
    "timestamp": "2025-01-15T10:30:00Z",
    "uptime_sec": 86400,
    "queue_depth": 0,
    "cameras_online": 2,
    "disk_usage_percent": 15.5,
    "config_version": "v3"
}

Heartbeat Response (Central → Edge): 

{
    "status": "ok",
    "pending_commands": [
        {
            "id": "cmd-uuid-123",
            "cmd": "config_update",
            "timestamp": "2025-01-15T10:30:00Z",
            "payload": { "cameras": [...] }
        }
    ]
}

Command Types: - config_update - Update collector configuration (cameras, upload settings) - reboot - Reboot the Raspberry Pi (with optional delay) - restart_service - Restart the collector service - force_upload - Immediately upload all buffered events

Command Acknowledgment (Edge → Central): 

POST /api/v1/commands/{cmd_id}/ack
{
    "status": "success",  # or "failed"
    "message": "Config applied successfully"
}

Edge Implementation: 

async def heartbeat_loop():
    """Send heartbeat every 60 seconds, process any pending commands."""
    while True:
        try:
            response = await client.post(
                f"{central_url}/api/v1/heartbeat",
                headers={"X-API-Key": config.api_key},
                json=get_health_status()
            )

            for cmd in response.json().get("pending_commands", []):
                try:
                    await handle_command(cmd)
                    await send_ack(cmd["id"], "success")
                except Exception as e:
                    logger.error(
                        "Command execution failed",
                        command_id=cmd["id"],
                        command_type=cmd.get("cmd"),
                        error_type=type(e).__name__,
                        error_message=str(e),
                        exc_info=True
                    )
                    await send_ack(cmd["id"], "failed", str(e))

        except ConnectionError as e:
            logger.warning(
                "Heartbeat failed, will retry",
                error_type=type(e).__name__,
                error_message=str(e),
                retry_interval_sec=60,
                central_url=central_url
            )

        await asyncio.sleep(60)

Central Health Monitoring: 

-- Collector status based on last heartbeat
SELECT
    id, name, last_heartbeat,
    CASE
        WHEN last_heartbeat > NOW() - INTERVAL '2 minutes' THEN 'online'
        WHEN last_heartbeat > NOW() - INTERVAL '10 minutes' THEN 'degraded'
        ELSE 'offline'
    END as status
FROM collectors;

Alerting Patterns

Pattern: Watchlist Matching

async def check_watchlist(plate_number: str) -> List[Alert]:
    """Check if plate matches any watchlist entries."""
    normalized = normalize_plate(plate_number)

    # Exact match
    matches = await db.query(
        Watchlist.plate_number == normalized
    ).all()

    # Partial match (if enabled)
    if settings.enable_partial_matching:
        partial = await db.query(
            Watchlist.plate_number.contains(normalized) |
            literal(normalized).contains(Watchlist.plate_number)
        ).all()
        matches.extend(partial)

    return [create_alert(m) for m in matches]

Pattern: User Alert Subscriptions

async def notify_subscribers(alert: Alert):
    """Send alert to all subscribed users."""
    subscribers = await db.query(
        AlertSubscription.watchlist_id == alert.watchlist_id,
        AlertSubscription.is_active == True
    ).all()

    for sub in subscribers:
        # WebSocket push
        await websocket_manager.send_to_user(
            sub.user_id,
            {"type": "alert", "data": alert.to_dict()}
        )

        # Future: Push notification to mobile

Mobile-Responsive UI Patterns

Pattern: Tailwind Breakpoints

<!-- Mobile-first design -->
<div class="
    flex flex-col          <!-- Mobile: stack vertically -->
    md:flex-row            <!-- Tablet+: horizontal -->
    lg:gap-8               <!-- Desktop: more spacing -->
">
    <div class="w-full md:w-1/2 lg:w-1/3">...</div>
    <div class="w-full md:w-1/2 lg:w-2/3">...</div>
</div>

<!-- Touch-friendly targets -->
<button class="
    p-4 min-h-[44px]       <!-- 44px minimum for touch -->
    text-base              <!-- Readable on mobile -->
    md:p-2 md:text-sm      <!-- Tighter on desktop -->
">
    Click Me
</button>

Pattern: Responsive Tables

<!-- Card view on mobile, table on desktop -->
<div class="hidden md:block">
    <table>...</table>
</div>
<div class="md:hidden">
    <div class="card">...</div>
    <div class="card">...</div>
</div>

Failure Handling Patterns

Edge Collector Failure Modes:

Scenario Behavior
Camera offline Adapter marks degraded, continues polling, auto-reconnects
Central server unreachable Buffer detections locally (up to 7 days), retry with backoff
Heartbeat fails Log warning, retry on next 60-second interval
Pi crash/reboot Systemd auto-restart, SQLite buffer survives reboot
Config error Reject invalid config, keep running with previous, send failed ack

Pattern: Exponential Backoff

async def upload_with_retry(detections: List[Detection], max_retries: int = 5):
    """Upload detections with exponential backoff."""
    for attempt in range(max_retries):
        try:
            await upload_detections(detections)
            return
        except ConnectionError as e:
            wait_time = min(300, 2 ** attempt * 30)  # 30s, 60s, 120s, 240s, 300s
            logger.warning(
                "Upload failed, retrying",
                retry_attempt=attempt + 1,
                max_retries=max_retries,
                retry_wait_sec=wait_time,
                detections_queued=len(detections),
                error_type=type(e).__name__,
                error_message=str(e)
            )
            await asyncio.sleep(wait_time)

    logger.error(
        "Max retries exceeded, detections remain in buffer",
        max_retries=max_retries,
        detections_in_buffer=len(detections)
    )

Last Updated: 2025-12-29 Maintained By: Development Team