Drift Detection

DriftDetector compares what django_migrations says ran against what actually exists in the PostgreSQL schema. It is entirely read-only — it never mutates the history table or the live schema.

Drift classification

Every drift incident is classified into one of three directions:

Foreign tables (tables present on this DB whose owning app routes to a different database) are listed separately and never auto-touched.

What is introspected

DriftDetector only checks operations it can introspect cheaply from the live schema:

• CreateModel — checks whether the table exists using information_schema.tables.
• AddField — checks whether the column exists using information_schema.columns. Resolves actual column names correctly: ForeignKey fields get an _id suffix, and explicit db_column overrides are respected.

All other operations (RunPython, RunSQL, RenameField, RemoveField, etc.) are trusted based on the django_migrations record. If a migration contains only non-introspectable operations, it is skipped by the drift scanner — no false positives.

Tri-state result

The core check returns one of three values:

Half-applied detection

Half-applied incidents are a special case: an unrecorded migration where the CreateModel table exists but subsequent AddField columns do not. This happens when:

• A CREATE TABLE statement ran (manually or from a partial migration run) but the ALTER TABLE ADD COLUMN statements that follow in the same migration file did not.
• The migration was never recorded in django_migrations.

Plain migrate would fail on this state because it would try to CREATE TABLE on an already-existing table. The repair strategy is to fake-apply the record so migrate sees the migration as done, then surface a warning that the missing columns need manual recovery.

Running a drift scan programmatically

from django_cfg.modules.django_migrator import Migrator, MigratorOptions
 
# Forensic check — no side effects
migrator = Migrator(options=MigratorOptions(dry_run=True))
report = migrator.check()
 
for db in report.db_reports:
    if db.drift and db.drift.has_issues():
        drift = db.drift
        print(f"{db.alias}: {drift.total_drift_count()} drift incidents")
 
        for inc in drift.recorded_missing:
            print(f"  RECORDED_MISSING  {inc.app_label}.{inc.migration_name}")
            print(f"    detail: {inc.detail}")
 
        for inc in drift.unrecorded_present:
            print(f"  UNRECORDED_PRESENT {inc.app_label}.{inc.migration_name}")
            print(f"    detail: {inc.detail}")
 
        for inc in drift.half_applied:
            print(f"  HALF_APPLIED      {inc.app_label}.{inc.migration_name}")
            print(f"    detail: {inc.detail}")
 
        for table in drift.foreign_tables:
            print(f"  FOREIGN TABLE: {table}  (app routes elsewhere)")

DriftReport data structure

@dataclass
class DriftReport:
    alias: str
    recorded_missing: list[DriftIncident]
    unrecorded_present: list[DriftIncident]
    half_applied: list[DriftIncident]
    foreign_tables: list[str]       # never auto-dropped
 
    def has_issues(self) -> bool: ...
    def total_drift_count(self) -> int: ...
 
@dataclass
class DriftIncident:
    app_label: str
    migration_name: str
    direction: Literal["recorded_missing", "unrecorded_present", "half_applied"]
    detail: str   # human-readable description of the specific mismatch

Common causes of drift