Skip to content

API Versioning & Database Migrations

Overview

This guide covers API versioning strategies and database migration best practices for AzmX backend systems using Django 5.2 with Django Ninja (invoicing-apis) and FastAPI (llms-embeddings-parsers).

API Versioning Strategies

Strategy Comparison

Strategy Pros Cons Use Case
URL Path Clear, cacheable, easy to route URL proliferation Public APIs
Header Clean URLs, flexible Hidden, harder to test Internal APIs
Query Param Simple, testable Not RESTful Testing

Recommendation: URL Path versioning (/api/v1/, /api/v2/)

Django Ninja Versioning

Multiple Version Support

# Create separate API instances
api_v1 = NinjaAPI(
    title="ZATCA API",
    version="1.0.0",
    openapi_url="/api/v1/openapi.json",
)

api_v2 = NinjaAPI(
    title="ZATCA API",
    version="2.0.0",
    openapi_url="/api/v2/openapi.json",
)

# urls.py
urlpatterns = [
    path('api/v1/', api_v1.urls),
    path('api/v2/', api_v2.urls),
]

Versioned Endpoints

# invoices/api_v1.py
@router.get("/", response=List[InvoiceSchemaV1])
def list_invoices(request):
    return Invoice.objects.all()

# invoices/api_v2.py
@router.get("/", response=List[InvoiceSchemaV2])
def list_invoices(request, include_metadata: bool = True):
    invoices = Invoice.objects.all()
    if include_metadata:
        invoices = invoices.select_related('seller', 'buyer')
    return invoices

FastAPI Versioning

Multiple Version Support

app = FastAPI(title="LLMs API")

# Version routers
app.include_router(
    anatomi_routes_v1.router,
    prefix="/api/v1/anatomi",
    tags=["anatomi-v1"]
)

app.include_router(
    anatomi_routes_v2.router,
    prefix="/api/v2/anatomi",
    tags=["anatomi-v2"]
)

Database Migration Workflow

graph TD
    A[Modify Models] --> B[Generate Migration]
    B --> C[Review Migration]
    C --> D{Complex?}
    D -->|Yes| E[Write Data Migration]
    D -->|No| F[Test Locally]
    E --> F
    F --> G[Commit & Deploy]
    G --> H{Issues?}
    H -->|Yes| I[Rollback]
    H -->|No| J[Complete]

Django Migration Commands

Essential Commands

# Generate migrations
python manage.py makemigrations
python manage.py makemigrations storage --name add_status_field

# Apply migrations
python manage.py migrate
python manage.py migrate storage 0005

# Check status
python manage.py showmigrations
python manage.py showmigrations storage

# Create empty migration for data operations
python manage.py makemigrations --empty storage --name populate_defaults

# Test migrations
python manage.py migrate --plan

Schema Migration Example

# storage/migrations/0001_draftinvoice.py
from django.db import migrations, models
import uuid

class Migration(migrations.Migration):
    dependencies = [
        ("storage", "trigram_migrations"),
    ]

    operations = [
        migrations.CreateModel(
            name="DraftInvoice",
            fields=[
                ("id", models.BigAutoField(primary_key=True)),
                ("uuid", models.UUIDField(default=uuid.uuid4, unique=True)),
                ("draft_data", models.JSONField()),
                ("created_at", models.DateTimeField(auto_now_add=True)),
            ],
        ),
    ]

Data Migration Example

# migrations/0006_populate_defaults.py
from django.db import migrations

def populate_defaults(apps, schema_editor):
    Invoice = apps.get_model('storage', 'Invoice')
    Invoice.objects.filter(status__isnull=True).update(status='draft')

def reverse_populate(apps, schema_editor):
    pass

class Migration(migrations.Migration):
    dependencies = [('storage', '0005_add_status_field')]
    operations = [
        migrations.RunPython(populate_defaults, reverse_populate),
    ]

Migration Best Practices

Adding Required Fields (3-Step Process)

# Step 1: Add as nullable
migrations.AddField('Invoice', 'status',
    models.CharField(max_length=20, null=True))

# Step 2: Populate data (separate migration)
def populate_status(apps, schema_editor):
    Invoice = apps.get_model('storage', 'Invoice')
    Invoice.objects.update(status='draft')

# Step 3: Make required (separate migration)
migrations.AlterField('Invoice', 'status',
    models.CharField(max_length=20, default='draft'))

Renaming Fields

# Use RenameField - preserves data
migrations.RenameField(
    model_name='invoice',
    old_name='amount',
    new_name='total_amount',
)

PostgreSQL-Specific Operations

from django.contrib.postgres.operations import TrigramExtension

class Migration(migrations.Migration):
    operations = [
        TrigramExtension(),
        migrations.RunSQL(
            "CREATE INDEX idx ON storage_invoice USING gin(to_tsvector('english', description));",
            "DROP INDEX idx;"
        ),
    ]

Slow Migrations on Large Tables

# Use raw SQL for performance
migrations.RunSQL(
    "UPDATE storage_invoice SET status = 'draft' WHERE status IS NULL;",
    reverse_sql="UPDATE storage_invoice SET status = NULL;"
)

# Or batch updates
def batch_update(apps, schema_editor):
    Invoice = apps.get_model('storage', 'Invoice')
    batch_size = 1000
    for i in range(0, Invoice.objects.count(), batch_size):
        batch = Invoice.objects.all()[i:i+batch_size]
        batch.update(status='draft')

Backward Compatibility

API Compatibility Patterns

Safe Changes (Non-Breaking): 

# Add optional fields
class InvoiceSchema(Schema):
    invoice_id: int
    amount: Decimal
    currency: str = "SAR"  # New optional field

# Add new endpoints
@router.post("/invoices/bulk/")
def bulk_create_invoices(request, invoices: List[InvoiceCreate]):
    pass

Deprecation Strategy: 

@router.get("/invoices/{id}/")
def get_invoice(request, id: int, legacy_format: bool = False):
    invoice = Invoice.objects.get(id=id)

    if legacy_format:
        return {"id": invoice.id, "amount": invoice.amount}

    return {
        "invoice_id": invoice.id,
        "total_amount": invoice.amount,
        "currency": invoice.currency,
    }

Database Compatibility

Column Renaming (Zero-Downtime): 

# Deploy 1: Add new column
migrations.AddField('Invoice', 'total_amount', DecimalField(null=True))

# Deploy 2: Copy data
def copy_data(apps, schema_editor):
    Invoice = apps.get_model('storage', 'Invoice')
    Invoice.objects.update(total_amount=F('amount'))

# Deploy 3: Remove old column (separate deployment)
migrations.RemoveField('Invoice', 'amount')

Table Renaming: 

# Create view for backward compatibility
migrations.RunSQL(
    "CREATE VIEW old_table AS SELECT * FROM new_table;",
    "DROP VIEW old_table;"
)

Rollback Procedures

Check Migration Status

# Show applied migrations
python manage.py showmigrations storage

# Output:
# storage
#  [X] 0001_initial
#  [X] 0002_add_status
#  [X] 0003_add_indexes

Rollback Commands

# Rollback to specific migration
python manage.py migrate storage 0002

# Rollback all migrations for app
python manage.py migrate storage zero

# Fake rollback (update state without SQL)
python manage.py migrate storage 0002 --fake

Emergency Rollback Process

# 1. Identify problem
python manage.py showmigrations storage --list

# 2. Rollback database
python manage.py migrate storage 0004_last_good_migration

# 3. Revert code
git revert <commit-hash>

# 4. Redeploy
systemctl restart gunicorn

Testing Migrations

# Test on fresh database
dropdb test_db && createdb test_db && python manage.py migrate

# Test with existing data
python manage.py loaddata fixtures/test_data.json && python manage.py migrate

CI/CD Checks: 

# .github/workflows/test.yml
- name: Check Migrations
  run: python manage.py makemigrations --check --dry-run

- name: Test Migrations
  run: python manage.py migrate && python manage.py test

Troubleshooting

Migration Conflicts

# Error: "Conflicting migrations detected"
python manage.py makemigrations --merge

Circular Dependencies

# Use string references
dependencies = [
    ('storage', '0005_invoice_status'),
    ('users', '0003_user_profile'),
]

# Or use run_before
run_before = [
    ('users', '0004_user_permissions'),
]

Inconsistent Migration State

# Check migration plan
python manage.py migrate --plan

# Fix by faking
python manage.py migrate --fake storage 0005
python manage.py migrate storage

Unapplied Migrations in Production

# Check what needs to be applied
python manage.py showmigrations | grep "\[ \]"

# Apply missing migrations
python manage.py migrate

Common Migration Patterns

Changing Field Type

# Step 1: Add new field
migrations.AddField('Invoice', 'amount_decimal', DecimalField(null=True))

# Step 2: Copy & convert data
def convert_amount(apps, schema_editor):
    Invoice = apps.get_model('storage', 'Invoice')
    for invoice in Invoice.objects.all():
        invoice.amount_decimal = Decimal(invoice.amount_int) / 100
        invoice.save()

# Step 3: Remove old field & rename
migrations.RemoveField('Invoice', 'amount_int')
migrations.RenameField('Invoice', 'amount_decimal', 'amount')

References