Django management commands are a good fit for repeatable, operational migration tasks: validating source files, importing data, reconciling results, retrying failed batches, and generating reports.
A typical command lives here:
myapp/
management/
commands/
import_customers.py
Run it with:
python manage.py import_customers /data/customers.csv --dry-run
1. Basic CSV import command
# myapp/management/commands/import_customers.py
import csv
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from customers.models import Customer
class Command(BaseCommand):
help = "Imports customers from a CSV export."
def add_arguments(self, parser):
parser.add_argument("csv_file", type=str)
parser.add_argument(
"--dry-run",
action="store_true",
help="Validate and process records, but roll back database changes.",
)
parser.add_argument(
"--batch-size",
type=int,
default=1000,
)
def handle(self, *args, **options):
csv_file = options["csv_file"]
dry_run = options["dry_run"]
batch_size = options["batch_size"]
try:
with open(csv_file, newline="", encoding="utf-8-sig") as file:
reader = csv.DictReader(file)
customers = []
for row_number, row in enumerate(reader, start=2):
customer = self.to_customer(row, row_number)
customers.append(customer)
if len(customers) >= batch_size:
self.save_batch(customers, dry_run)
customers = []
if customers:
self.save_batch(customers, dry_run)
except FileNotFoundError:
raise CommandError(f"File not found: {csv_file}")
self.stdout.write(self.style.SUCCESS("Import completed successfully."))
def to_customer(self, row, row_number):
external_id = row.get("CustomerID", "").strip()
email = row.get("Email", "").strip().lower()
if not external_id:
raise CommandError(f"Row {row_number}: CustomerID is required.")
if not email:
raise CommandError(f"Row {row_number}: Email is required.")
return Customer(
legacy_id=external_id,
email=email,
first_name=row.get("FirstName", "").strip(),
last_name=row.get("LastName", "").strip(),
)
def save_batch(self, customers, dry_run):
with transaction.atomic():
Customer.objects.bulk_create(
customers,
batch_size=len(customers),
ignore_conflicts=True,
)
if dry_run:
transaction.set_rollback(True)
self.stdout.write(
f"Validated {len(customers)} records (dry run)."
)
else:
self.stdout.write(f"Imported {len(customers)} records.")
Key interview points:
add_arguments() makes the operation configurable and automation-friendly.--dry-run lets operators validate real input without persisting changes.transaction.atomic() prevents partial writes within a batch.bulk_create() avoids one insert per ORM object.ignore_conflicts=True can help make reruns safer, but it must be used deliberately because it can hide unexpected conflicts.
2. Upsert-style migration using a stable legacy identifier
Migrations often need to be rerunnable. A legacy system’s primary key or external ID should generally be stored and enforced as unique in the destination system.
# models.py
class Customer(models.Model):
legacy_id = models.CharField(max_length=100, unique=True)
email = models.EmailField()
first_name = models.CharField(max_length=100)
last_name = models.CharField(max_length=100)
Then use update_or_create for smaller imports:
from django.core.management.base import BaseCommand
from django.db import transaction
from customers.models import Customer
class Command(BaseCommand):
help = "Synchronizes customers from a source system."
def handle(self, *args, **options):
source_rows = self.fetch_legacy_customers()
created = 0
updated = 0
with transaction.atomic():
for row in source_rows:
customer, was_created = Customer.objects.update_or_create(
legacy_id=row["id"],
defaults={
"email": row["email"].lower().strip(),
"first_name": row["first_name"].strip(),
"last_name": row["last_name"].strip(),
},
)
if was_created:
created += 1
else:
updated += 1
self.stdout.write(
self.style.SUCCESS(
f"Migration complete. Created={created}, Updated={updated}"
)
)
def fetch_legacy_customers(self):
# Could read XML, call an API, query a legacy DB, etc.
return [
{
"id": "LEGACY-1001",
"email": "alice@example.com",
"first_name": "Alice",
"last_name": "Smith",
}
]
For very large datasets, update_or_create() can be too slow because it performs database work per record. A senior-level answer should acknowledge that you would usually:
- Fetch existing IDs in chunks.
- Split incoming records into creates vs. updates.
- Use
bulk_create() and bulk_update().
3. Efficient bulk create/update pattern
from django.core.management.base import BaseCommand
from django.db import transaction
from customers.models import Customer
class Command(BaseCommand):
help = "Bulk synchronizes customers efficiently."
def handle(self, *args, **options):
rows = self.fetch_rows()
legacy_ids = [row["id"] for row in rows]
existing_customers = Customer.objects.in_bulk(
legacy_ids,
field_name="legacy_id",
)
to_create = []
to_update = []
for row in rows:
legacy_id = row["id"]
if legacy_id in existing_customers:
customer = existing_customers[legacy_id]
customer.email = row["email"].strip().lower()
customer.first_name = row["first_name"].strip()
customer.last_name = row["last_name"].strip()
to_update.append(customer)
else:
to_create.append(
Customer(
legacy_id=legacy_id,
email=row["email"].strip().lower(),
first_name=row["first_name"].strip(),
last_name=row["last_name"].strip(),
)
)
with transaction.atomic():
Customer.objects.bulk_create(to_create, batch_size=1000)
Customer.objects.bulk_update(
to_update,
fields=["email", "first_name", "last_name"],
batch_size=1000,
)
self.stdout.write(
self.style.SUCCESS(
f"Created {len(to_create)}; updated {len(to_update)}."
)
)
def fetch_rows(self):
return []
This is commonly preferable for high-volume migrations, but you need to be aware that Django bulk operations:
- Do not call
save().
- Do not emit
pre_save / post_save signals.
- May not populate all generated IDs consistently across every database/version combination.
- Need explicit validation before persistence.
4. XML parsing and data mapping
Legacy systems often expose XML with inconsistent or older schemas.
import xml.etree.ElementTree as ET
from django.core.management.base import BaseCommand, CommandError
from migration.services.customer_import import CustomerImportService
class Command(BaseCommand):
help = "Imports customers from a legacy XML export."
def add_arguments(self, parser):
parser.add_argument("xml_file")
parser.add_argument("--organization-id", type=int, required=True)
def handle(self, *args, **options):
try:
tree = ET.parse(options["xml_file"])
except ET.ParseError as exc:
raise CommandError(f"Invalid XML: {exc}")
root = tree.getroot()
records = []
for node in root.findall("./Customers/Customer"):
records.append(
{
"legacy_id": self.required_text(node, "CustomerNumber"),
"email": self.required_text(node, "Email").lower(),
"first_name": self.optional_text(node, "FirstName"),
"last_name": self.optional_text(node, "LastName"),
"status": self.map_status(self.optional_text(node, "Status")),
}
)
service = CustomerImportService(
organization_id=options["organization_id"],
)
result = service.import_records(records)
self.stdout.write(
self.style.SUCCESS(
f"Created={result.created}, updated={result.updated}, "
f"failed={result.failed}"
)
)
def required_text(self, parent, tag_name):
value = self.optional_text(parent, tag_name)
if not value:
raise CommandError(f"Missing required XML field: {tag_name}")
return value
def optional_text(self, parent, tag_name):
element = parent.find(tag_name)
return element.text.strip() if element is not None and element.text else ""
def map_status(self, legacy_status):
mapping = {
"A": "active",
"I": "inactive",
"S": "suspended",
}
try:
return mapping[legacy_status]
except KeyError:
raise CommandError(f"Unknown legacy status: {legacy_status}")
For untrusted XML, use a hardened parser such as defusedxml rather than the standard library parser, to reduce XML entity-expansion risks.
5. Separate command orchestration from migration business logic
A strong design keeps the command thin. The command should parse arguments, establish dependencies, report results, and set an exit status. Mapping, validation, persistence, and retry behavior belong in a testable service.
# migration/services/customer_import.py
from dataclasses import dataclass
from django.db import transaction
from customers.models import Customer
@dataclass
class ImportResult:
created: int = 0
updated: int = 0
failed: int = 0
class CustomerImportService:
def __init__(self, organization_id):
self.organization_id = organization_id
def import_records(self, records):
result = ImportResult()
for batch in self.batches(records, size=500):
self.import_batch(batch, result)
return result
@transaction.atomic
def import_batch(self, batch, result):
for record in batch:
try:
_, created = Customer.objects.update_or_create(
organization_id=self.organization_id,
legacy_id=record["legacy_id"],
defaults={
"email": record["email"],
"first_name": record["first_name"],
"last_name": record["last_name"],
"status": record["status"],
},
)
except Exception:
# In production, catch expected exceptions more narrowly,
# record structured failure information, and continue or stop
# based on the migration policy.
result.failed += 1
continue
if created:
result.created += 1
else:
result.updated += 1
@staticmethod
def batches(records, size):
for index in range(0, len(records), size):
yield records[index:index + size]
6. Checkpointed migration jobs
Long-running migrations should be resumable. A common approach is to persist job state and a checkpoint, such as the last processed source ID, row number, or cursor.
# models.py
class MigrationJob(models.Model):
class Status(models.TextChoices):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
organization_id = models.IntegerField()
status = models.CharField(max_length=20, choices=Status.choices)
last_legacy_id = models.CharField(max_length=100, blank=True)
processed_count = models.PositiveIntegerField(default=0)
failed_count = models.PositiveIntegerField(default=0)
error_message = models.TextField(blank=True)
# management/commands/run_migration_job.py
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from migration.models import MigrationJob
from migration.services.source_client import LegacySourceClient
from migration.services.customer_import import CustomerImportService
class Command(BaseCommand):
help = "Runs or resumes a customer migration job."
def add_arguments(self, parser):
parser.add_argument("job_id", type=int)
parser.add_argument("--page-size", type=int, default=500)
def handle(self, *args, **options):
try:
job = MigrationJob.objects.get(pk=options["job_id"])
except MigrationJob.DoesNotExist:
raise CommandError("Migration job does not exist.")
job.status = MigrationJob.Status.RUNNING
job.save(update_fields=["status"])
client = LegacySourceClient(job.organization_id)
importer = CustomerImportService(job.organization_id)
try:
while True:
records = client.fetch_customers(
after_id=job.last_legacy_id,
limit=options["page_size"],
)
if not records:
break
with transaction.atomic():
result = importer.import_records(records)
job.last_legacy_id = records[-1]["legacy_id"]
job.processed_count += len(records)
job.failed_count += result.failed
job.save(
update_fields=[
"last_legacy_id",
"processed_count",
"failed_count",
]
)
job.status = MigrationJob.Status.COMPLETED
job.save(update_fields=["status"])
self.stdout.write(self.style.SUCCESS("Migration completed."))
except Exception as exc:
job.status = MigrationJob.Status.FAILED
job.error_message = str(exc)
job.save(update_fields=["status", "error_message"])
raise CommandError(f"Migration failed: {exc}")
Important nuance: update the checkpoint in the same transaction as the imported batch. Otherwise, a crash can produce either:
- data committed but checkpoint not updated → duplicate processing on retry, or
- checkpoint updated but data not committed → skipped records.
Idempotent writes plus transactional checkpoints make retries safe.
7. Validation-only command
It is often valuable to validate files before the actual import window.
from django.core.management.base import BaseCommand, CommandError
from migration.services.csv_validator import CustomerCsvValidator
class Command(BaseCommand):
help = "Validates a customer migration CSV without importing it."
def add_arguments(self, parser):
parser.add_argument("csv_file")
parser.add_argument("--error-limit", type=int, default=100)
def handle(self, *args, **options):
validator = CustomerCsvValidator(
error_limit=options["error_limit"]
)
report = validator.validate_file(options["csv_file"])
self.stdout.write(f"Rows checked: {report.total_rows}")
self.stdout.write(f"Valid rows: {report.valid_rows}")
self.stdout.write(f"Invalid rows: {report.invalid_rows}")
for error in report.errors:
self.stderr.write(
f"Row {error.row_number}, field '{error.field}': {error.message}"
)
if report.invalid_rows:
raise CommandError("Validation failed.")
self.stdout.write(self.style.SUCCESS("Validation passed."))
Validation can check:
- Required columns and duplicate headers.
- Required values and type/format constraints.
- Referential integrity, such as whether a customer’s organization exists.
- Controlled vocabulary mappings.
- Duplicate legacy IDs within the source file.
- Conflicts with existing destination records.
- Cross-field rules, such as “inactive date is required when status is inactive.”
8. Reconciliation command
A migration is not complete just because the import did not crash. Reconciliation checks source and target totals, missing IDs, and key field discrepancies.
from django.core.management.base import BaseCommand
from customers.models import Customer
from migration.services.source_client import LegacySourceClient
class Command(BaseCommand):
help = "Compares legacy customer IDs against migrated customer IDs."
def add_arguments(self, parser):
parser.add_argument("--organization-id", type=int, required=True)
def handle(self, *args, **options):
organization_id = options["organization_id"]
source_ids = set(
LegacySourceClient(organization_id).get_all_customer_ids()
)
target_ids = set(
Customer.objects.filter(
organization_id=organization_id
).values_list("legacy_id", flat=True)
)
missing_in_target = source_ids - target_ids
unexpected_in_target = target_ids - source_ids
self.stdout.write(f"Source count: {len(source_ids)}")
self.stdout.write(f"Target count: {len(target_ids)}")
self.stdout.write(f"Missing in target: {len(missing_in_target)}")
self.stdout.write(f"Unexpected in target: {len(unexpected_in_target)}")
if missing_in_target:
self.stderr.write(
"Example missing IDs: "
+ ", ".join(sorted(missing_in_target)[:20])
)
For millions of records, avoid loading IDs into Python sets. Use database-side staging tables, sorted-stream comparisons, or chunked reconciliation.
What interviewers are usually looking for
When discussing Django migration commands, emphasize that production migration work needs more than parsing and inserting:
- Idempotency: reruns should not corrupt or duplicate records.
- Atomicity: define transaction boundaries at a sensible batch size.
- Validation: fail early on schema and data-quality problems.
- Observability: structured logs, counts, job IDs, timings, and actionable error reports.
- Resumability: durable checkpoints and clear job state.
- Performance: chunked reads, bulk inserts/updates, appropriate indexes, and avoiding N+1 queries.
- Safety: dry runs, staging environments, backups, rollback strategy, and phased/canary migrations.
- Reconciliation: prove migrated data is complete and correct after loading.
- Operational usability: clear command options, non-zero exit codes on failure, and commands that can run from CI, scheduled jobs, or operational runbooks.