Adding migrations to Django apps requires understanding the migration system's integration with app structure, handling initial migrations, and managing migrations across different app states. This section covers best practices for incorporating migrations into new and existing applications.
# Setting up migrations for a new app
# 1. Create the app structure
"""
blog/
├── __init__.py
├── admin.py
├── apps.py
├── models.py
├── tests.py
├── views.py
└── migrations/
└── __init__.py
"""
# 2. Define initial models
# blog/models.py
from django.db import models
from django.contrib.auth.models import User
class Category(models.Model):
name = models.CharField(max_length=100, unique=True)
description = models.TextField(blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
verbose_name_plural = "categories"
ordering = ['name']
def __str__(self):
return self.name
class Tag(models.Model):
name = models.CharField(max_length=50, unique=True)
slug = models.SlugField(unique=True)
def __str__(self):
return self.name
class Post(models.Model):
STATUS_CHOICES = [
('draft', 'Draft'),
('published', 'Published'),
('archived', 'Archived'),
]
title = models.CharField(max_length=200)
slug = models.SlugField(unique=True)
content = models.TextField()
excerpt = models.TextField(blank=True)
author = models.ForeignKey(User, on_delete=models.CASCADE)
category = models.ForeignKey(Category, on_delete=models.CASCADE)
tags = models.ManyToManyField(Tag, blank=True)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='draft')
featured = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
published_at = models.DateTimeField(null=True, blank=True)
class Meta:
ordering = ['-created_at']
indexes = [
models.Index(fields=['status', 'published_at']),
models.Index(fields=['category', 'status']),
models.Index(fields=['author', 'status']),
]
def __str__(self):
return self.title
# 3. Generate initial migration
# python manage.py makemigrations blog
# Generated migration: blog/migrations/0001_initial.py
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='Category',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100, unique=True)),
('description', models.TextField(blank=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
],
options={
'verbose_name_plural': 'categories',
'ordering': ['name'],
},
),
migrations.CreateModel(
name='Tag',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=50, unique=True)),
('slug', models.SlugField(unique=True)),
],
),
migrations.CreateModel(
name='Post',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('title', models.CharField(max_length=200)),
('slug', models.SlugField(unique=True)),
('content', models.TextField()),
('excerpt', models.TextField(blank=True)),
('status', models.CharField(choices=[('draft', 'Draft'), ('published', 'Published'), ('archived', 'Archived')], default='draft', max_length=20)),
('featured', models.BooleanField(default=False)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('published_at', models.DateTimeField(blank=True, null=True)),
('author', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
('category', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='blog.category')),
('tags', models.ManyToManyField(blank=True, to='blog.tag')),
],
options={
'ordering': ['-created_at'],
},
),
migrations.AddIndex(
model_name='post',
index=models.Index(fields=['status', 'published_at'], name='blog_post_status_5b7c8a_idx'),
),
migrations.AddIndex(
model_name='post',
index=models.Index(fields=['category', 'status'], name='blog_post_categor_9e5b8f_idx'),
),
migrations.AddIndex(
model_name='post',
index=models.Index(fields=['author', 'status'], name='blog_post_author_0c6e9d_idx'),
),
]
# blog/apps.py - Proper app configuration
from django.apps import AppConfig
class BlogConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'blog'
verbose_name = 'Blog Management'
def ready(self):
"""Initialize app when Django starts"""
# Import signals, register admin, etc.
import blog.signals # noqa
# Custom migration operations for app-specific needs
class AppSpecificMigrationOperations:
"""Custom operations for blog app migrations"""
@staticmethod
def create_default_categories(apps, schema_editor):
"""Create default categories for the blog"""
Category = apps.get_model('blog', 'Category')
default_categories = [
{
'name': 'Technology',
'description': 'Posts about technology and programming'
},
{
'name': 'Lifestyle',
'description': 'Posts about lifestyle and personal experiences'
},
{
'name': 'Business',
'description': 'Posts about business and entrepreneurship'
},
]
for category_data in default_categories:
Category.objects.get_or_create(
name=category_data['name'],
defaults={'description': category_data['description']}
)
@staticmethod
def create_default_tags(apps, schema_editor):
"""Create default tags for the blog"""
Tag = apps.get_model('blog', 'Tag')
default_tags = [
{'name': 'Django', 'slug': 'django'},
{'name': 'Python', 'slug': 'python'},
{'name': 'Web Development', 'slug': 'web-development'},
{'name': 'Tutorial', 'slug': 'tutorial'},
{'name': 'Tips', 'slug': 'tips'},
]
for tag_data in default_tags:
Tag.objects.get_or_create(
name=tag_data['name'],
defaults={'slug': tag_data['slug']}
)
@staticmethod
def setup_initial_data(apps, schema_editor):
"""Set up initial data for the blog app"""
# Create default categories and tags
AppSpecificMigrationOperations.create_default_categories(apps, schema_editor)
AppSpecificMigrationOperations.create_default_tags(apps, schema_editor)
# Create sample content if in development
from django.conf import settings
if settings.DEBUG:
AppSpecificMigrationOperations.create_sample_content(apps, schema_editor)
@staticmethod
def create_sample_content(apps, schema_editor):
"""Create sample content for development"""
from django.contrib.auth import get_user_model
User = get_user_model()
Post = apps.get_model('blog', 'Post')
Category = apps.get_model('blog', 'Category')
Tag = apps.get_model('blog', 'Tag')
# Get or create a sample user
sample_user, created = User.objects.get_or_create(
username='admin',
defaults={
'email': 'admin@example.com',
'first_name': 'Admin',
'last_name': 'User',
'is_staff': True,
'is_superuser': True,
}
)
# Create sample posts
tech_category = Category.objects.get(name='Technology')
django_tag = Tag.objects.get(name='Django')
python_tag = Tag.objects.get(name='Python')
sample_posts = [
{
'title': 'Getting Started with Django',
'slug': 'getting-started-with-django',
'content': 'This is a comprehensive guide to getting started with Django...',
'excerpt': 'Learn the basics of Django web framework',
'category': tech_category,
'author': sample_user,
'status': 'published',
'tags': [django_tag, python_tag],
},
{
'title': 'Advanced Django Patterns',
'slug': 'advanced-django-patterns',
'content': 'Explore advanced patterns and best practices in Django...',
'excerpt': 'Advanced techniques for Django developers',
'category': tech_category,
'author': sample_user,
'status': 'draft',
'tags': [django_tag],
},
]
for post_data in sample_posts:
tags = post_data.pop('tags', [])
post, created = Post.objects.get_or_create(
slug=post_data['slug'],
defaults=post_data
)
if created and tags:
post.tags.set(tags)
# Migration with initial data setup
class Migration(migrations.Migration):
"""Migration with initial data setup"""
dependencies = [
('blog', '0001_initial'),
]
operations = [
migrations.RunPython(
code=AppSpecificMigrationOperations.setup_initial_data,
reverse_code=migrations.RunPython.noop,
),
]
# Scenario: Adding migrations to an app that was created without them
class LegacyAppMigrationSetup:
"""Setup migrations for legacy apps"""
@staticmethod
def create_initial_migration_for_existing_app():
"""Create initial migration for app with existing database tables"""
# Step 1: Create migrations directory if it doesn't exist
import os
from django.apps import apps
app_config = apps.get_app_config('legacy_app')
migrations_dir = os.path.join(app_config.path, 'migrations')
if not os.path.exists(migrations_dir):
os.makedirs(migrations_dir)
# Create __init__.py
init_file = os.path.join(migrations_dir, '__init__.py')
with open(init_file, 'w') as f:
f.write('')
# Step 2: Generate initial migration
# python manage.py makemigrations legacy_app
# Step 3: Fake apply the initial migration since tables already exist
# python manage.py migrate legacy_app 0001 --fake-initial
return migrations_dir
@staticmethod
def handle_existing_data_conflicts():
"""Handle conflicts when adding migrations to apps with existing data"""
# Common issues and solutions:
# 1. Unique constraint violations
def fix_unique_constraints(apps, schema_editor):
"""Fix unique constraint violations in existing data"""
LegacyModel = apps.get_model('legacy_app', 'LegacyModel')
# Find and fix duplicate entries
duplicates = LegacyModel.objects.values('email').annotate(
count=Count('email')
).filter(count__gt=1)
for duplicate in duplicates:
# Keep the first entry, remove others
entries = LegacyModel.objects.filter(email=duplicate['email'])
entries.exclude(id=entries.first().id).delete()
# 2. Foreign key constraint violations
def fix_foreign_key_constraints(apps, schema_editor):
"""Fix foreign key constraint violations"""
LegacyModel = apps.get_model('legacy_app', 'LegacyModel')
RelatedModel = apps.get_model('legacy_app', 'RelatedModel')
# Find orphaned records
orphaned = LegacyModel.objects.filter(
related_id__isnull=False
).exclude(
related_id__in=RelatedModel.objects.values_list('id', flat=True)
)
# Option 1: Set to NULL if field allows it
orphaned.update(related_id=None)
# Option 2: Create placeholder records
# for record in orphaned:
# RelatedModel.objects.get_or_create(id=record.related_id)
# 3. NOT NULL constraint violations
def fix_not_null_constraints(apps, schema_editor):
"""Fix NOT NULL constraint violations"""
LegacyModel = apps.get_model('legacy_app', 'LegacyModel')
# Set default values for NULL fields that will become NOT NULL
LegacyModel.objects.filter(required_field__isnull=True).update(
required_field='default_value'
)
return {
'unique_constraints': fix_unique_constraints,
'foreign_key_constraints': fix_foreign_key_constraints,
'not_null_constraints': fix_not_null_constraints,
}
# Migration for legacy app setup
class LegacyAppInitialMigration(migrations.Migration):
"""Initial migration for legacy app with data cleanup"""
initial = True
dependencies = []
operations = [
# First, create the models as they should be
migrations.CreateModel(
name='LegacyModel',
fields=[
('id', models.AutoField(primary_key=True)),
('name', models.CharField(max_length=100)),
('email', models.EmailField(unique=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
],
),
# Then clean up existing data to match new constraints
migrations.RunPython(
code=LegacyAppMigrationSetup.handle_existing_data_conflicts()['unique_constraints'],
reverse_code=migrations.RunPython.noop,
),
]
# Handling schema differences in legacy apps
class SchemaReconciliation:
"""Reconcile differences between model definitions and existing schema"""
@staticmethod
def analyze_schema_differences():
"""Analyze differences between models and database schema"""
from django.db import connection
from django.apps import apps
differences = {}
with connection.cursor() as cursor:
# Get existing table structure
cursor.execute("""
SELECT table_name, column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_schema = %s
""", [connection.settings_dict['NAME']])
existing_schema = {}
for row in cursor.fetchall():
table_name, column_name, data_type, is_nullable = row
if table_name not in existing_schema:
existing_schema[table_name] = {}
existing_schema[table_name][column_name] = {
'type': data_type,
'nullable': is_nullable == 'YES'
}
# Compare with model definitions
for app_config in apps.get_app_configs():
for model in app_config.get_models():
table_name = model._meta.db_table
if table_name in existing_schema:
model_fields = {}
for field in model._meta.fields:
model_fields[field.column] = {
'type': field.db_type(connection),
'nullable': field.null
}
# Find differences
table_differences = []
for field_name, field_info in model_fields.items():
if field_name not in existing_schema[table_name]:
table_differences.append({
'type': 'missing_column',
'field': field_name,
'expected': field_info
})
else:
existing_field = existing_schema[table_name][field_name]
if existing_field['nullable'] != field_info['nullable']:
table_differences.append({
'type': 'nullable_mismatch',
'field': field_name,
'existing': existing_field['nullable'],
'expected': field_info['nullable']
})
if table_differences:
differences[table_name] = table_differences
return differences
@staticmethod
def create_reconciliation_migration(differences):
"""Create migration to reconcile schema differences"""
operations = []
for table_name, table_differences in differences.items():
for difference in table_differences:
if difference['type'] == 'missing_column':
# Add missing column
operations.append(
migrations.RunSQL(
sql=f"ALTER TABLE {table_name} ADD COLUMN {difference['field']} ...",
reverse_sql=f"ALTER TABLE {table_name} DROP COLUMN {difference['field']}"
)
)
elif difference['type'] == 'nullable_mismatch':
# Fix nullable constraint
if difference['expected']:
# Make nullable
operations.append(
migrations.RunSQL(
sql=f"ALTER TABLE {table_name} ALTER COLUMN {difference['field']} DROP NOT NULL",
reverse_sql=f"ALTER TABLE {table_name} ALTER COLUMN {difference['field']} SET NOT NULL"
)
)
else:
# Make NOT NULL (need to handle existing NULL values first)
operations.extend([
migrations.RunSQL(
sql=f"UPDATE {table_name} SET {difference['field']} = 'default' WHERE {difference['field']} IS NULL",
reverse_sql=migrations.RunSQL.noop
),
migrations.RunSQL(
sql=f"ALTER TABLE {table_name} ALTER COLUMN {difference['field']} SET NOT NULL",
reverse_sql=f"ALTER TABLE {table_name} ALTER COLUMN {difference['field']} DROP NOT NULL"
)
])
return operations
# Extending third-party apps with custom migrations
class ThirdPartyAppExtension:
"""Extend third-party apps with custom migrations"""
@staticmethod
def extend_user_model():
"""Extend Django's User model with additional fields"""
# Create a migration in your app that extends auth.User
# This is safer than modifying third-party app migrations directly
class Migration(migrations.Migration):
dependencies = [
('auth', '0012_alter_user_first_name_max_length'),
('your_app', '0001_initial'),
]
operations = [
migrations.CreateModel(
name='UserProfile',
fields=[
('id', models.AutoField(primary_key=True)),
('user', models.OneToOneField(
'auth.User',
on_delete=models.CASCADE,
related_name='profile'
)),
('bio', models.TextField(blank=True)),
('avatar', models.ImageField(upload_to='avatars/', blank=True)),
('birth_date', models.DateField(null=True, blank=True)),
],
),
]
@staticmethod
def customize_third_party_model():
"""Add custom fields to third-party app models"""
# Example: Adding fields to django-allauth models
class Migration(migrations.Migration):
dependencies = [
('account', '0003_emailaddress_idx_upper_email'),
('your_app', '0002_userprofile'),
]
operations = [
# Add custom field to EmailAddress model
migrations.AddField(
model_name='emailaddress',
name='is_marketing_enabled',
field=models.BooleanField(default=False),
),
# Add index for the new field
migrations.AddIndex(
model_name='emailaddress',
index=models.Index(
fields=['is_marketing_enabled'],
name='account_emailaddress_marketing_idx'
),
),
]
@staticmethod
def create_proxy_model_migration():
"""Create proxy models for third-party apps"""
class Migration(migrations.Migration):
dependencies = [
('auth', '0012_alter_user_first_name_max_length'),
('your_app', '0003_extend_emailaddress'),
]
operations = [
migrations.CreateModel(
name='CustomUser',
fields=[],
options={
'proxy': True,
},
bases=('auth.user',),
managers=[
('objects', django.contrib.auth.models.UserManager()),
],
),
]
# Managing third-party app migration dependencies
class ThirdPartyDependencyManager:
"""Manage dependencies on third-party app migrations"""
@staticmethod
def check_third_party_migration_status():
"""Check status of third-party app migrations"""
from django.db.migrations.loader import MigrationLoader
loader = MigrationLoader(connection)
third_party_apps = [
'allauth', 'rest_framework', 'corsheaders',
'django_extensions', 'debug_toolbar'
]
status = {}
for app_label in third_party_apps:
if app_label in loader.migrated_apps:
app_migrations = []
for migration_key in loader.graph.nodes:
if migration_key[0] == app_label:
is_applied = migration_key in loader.applied_migrations
app_migrations.append({
'name': migration_key[1],
'applied': is_applied
})
status[app_label] = {
'installed': True,
'migrations': app_migrations,
'all_applied': all(m['applied'] for m in app_migrations)
}
else:
status[app_label] = {
'installed': False,
'migrations': [],
'all_applied': False
}
return status
@staticmethod
def create_safe_third_party_dependency():
"""Create safe dependency on third-party app migration"""
# Instead of depending on specific migration numbers,
# depend on the app's latest migration at time of creation
from django.db.migrations.loader import MigrationLoader
def get_latest_migration(app_label):
loader = MigrationLoader(connection)
if app_label not in loader.migrated_apps:
return None
# Get leaf nodes for the app
leaf_nodes = [
node for node in loader.graph.leaf_nodes()
if node[0] == app_label
]
return leaf_nodes[0] if leaf_nodes else None
# Example usage in migration
latest_auth_migration = get_latest_migration('auth')
class Migration(migrations.Migration):
dependencies = [
latest_auth_migration, # Safe dependency
('your_app', '0001_initial'),
]
operations = [
# Your operations here
]
@staticmethod
def handle_missing_third_party_migrations():
"""Handle cases where third-party migrations are missing"""
def conditional_operation(apps, schema_editor):
"""Operation that only runs if third-party app is available"""
try:
# Try to get the third-party model
ThirdPartyModel = apps.get_model('third_party_app', 'SomeModel')
# Perform operation if model exists
# ... your code here ...
except LookupError:
# Third-party app not installed or model doesn't exist
print("Third-party app not available, skipping operation")
return conditional_operation
# App-specific migration utilities
class AppMigrationUtilities:
"""Utilities for app-specific migration management"""
@staticmethod
def create_app_migration_template(app_name, migration_name):
"""Create a template for app-specific migrations"""
template = f'''
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
"""
{migration_name} for {app_name} app
Description: Add description of what this migration does
Safety: [SAFE/CAUTION/DANGEROUS]
- Explain any potential risks
- Note any data loss possibilities
- Mention performance implications
Rollback: [SAFE/MANUAL/IMPOSSIBLE]
- Explain rollback implications
"""
dependencies = [
('{app_name}', 'XXXX_previous_migration'),
]
operations = [
# Add your operations here
]
# Set to False for operations that can't run in transactions
atomic = True
'''
return template
@staticmethod
def validate_app_migration_structure(app_name):
"""Validate migration structure for an app"""
import os
from django.apps import apps
app_config = apps.get_app_config(app_name)
migrations_dir = os.path.join(app_config.path, 'migrations')
validation_results = {
'valid': True,
'issues': [],
'recommendations': []
}
# Check if migrations directory exists
if not os.path.exists(migrations_dir):
validation_results['valid'] = False
validation_results['issues'].append(
f"Migrations directory missing for {app_name}"
)
validation_results['recommendations'].append(
f"Run: python manage.py makemigrations {app_name}"
)
# Check for __init__.py
init_file = os.path.join(migrations_dir, '__init__.py')
if not os.path.exists(init_file):
validation_results['issues'].append(
f"Missing __init__.py in {app_name}/migrations/"
)
validation_results['recommendations'].append(
f"Create empty __init__.py file in {migrations_dir}"
)
# Check migration file naming
if os.path.exists(migrations_dir):
migration_files = [
f for f in os.listdir(migrations_dir)
if f.endswith('.py') and not f.startswith('__')
]
for migration_file in migration_files:
if not migration_file.startswith('0'):
validation_results['issues'].append(
f"Migration file {migration_file} doesn't follow naming convention"
)
return validation_results
Adding migrations to Django apps requires careful consideration of existing data, dependencies, and app structure. Proper setup ensures smooth schema evolution and maintains data integrity across different deployment environments.
Transaction Handling
Django migrations run within database transactions by default, providing atomicity and consistency during schema changes. Understanding transaction behavior in migrations is crucial for maintaining data integrity and handling complex migration scenarios safely.
Reversing Migrations
Migration reversal is a critical aspect of Django's migration system, allowing you to undo database changes safely. Understanding how to reverse migrations, handle data preservation, and manage rollback scenarios is essential for maintaining database integrity during development and production deployments.