#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import shutil import datetime from pathlib import Path from typing import List, Optional, Set, Dict, Tuple import subprocess from django.conf import settings from django.apps import apps from django.core.management import BaseCommand, call_command, CommandError from django.db import connections, DEFAULT_DB_ALIAS, transaction from django.db.migrations.executor import MigrationExecutor # ----------------------------- # Progress helpers (no deps) # ----------------------------- class Progress: def __init__(self, enabled: bool = True, mode: str = "bar", width: int = 36, stream=None): self.enabled = enabled self.mode = mode # "bar" | "dots" | "none" self.width = width self.stream = stream self.total = 0 self.current = 0 self.label = "" def start(self, total: int, label: str = ""): self.total = max(int(total), 0) self.current = 0 self.label = label if not self.enabled or self.mode == "none": return if self.mode == "bar": self._render() elif self.mode == "dots": self._write(f"{label} ") def step(self, inc: int = 1): if self.total <= 0: return self.current = min(self.total, self.current + inc) if not self.enabled or self.mode == "none": return if self.mode == "bar": self._render() elif self.mode == "dots": self._write(".") def done(self, suffix: str = " OK"): if not self.enabled or self.mode == "none": return if self.mode == "bar": # ensure full self.current = self.total self._render(final=True) self._write("\n") elif self.mode == "dots": self._write(suffix + "\n") def _render(self, final: bool = False): # bar + % + counts pct = int((self.current / self.total) * 100) if self.total else 100 filled = int((self.current / self.total) * self.width) if self.total else self.width bar = "█" * filled + "░" * (self.width - filled) line = f"\r{self.label} [{bar}] {pct:3d}% ({self.current}/{self.total})" self._write(line) def _write(self, s: str): if self.stream is None: # fallback to stdout via print without newline control print(s, end="", flush=True) else: self.stream.write(s) self.stream.flush() class Command(BaseCommand): help = ( "DANGEROUS: wipe Django migration files + django_migrations, then rebuild and fake-migrate.\n" "Use only on DEV/TEST environments. Requires --force." ) def _run_cmd(self, cmd: List[str], cwd: Optional[str] = None) -> Tuple[int, str, str]: p = subprocess.Popen( cmd, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) out, err = p.communicate() return p.returncode, out.strip(), err.strip() def _git_commit(self, message: str, allow_dirty: bool, noinput: bool) -> None: """ Safer behavior: - by default, refuse if repo is dirty (unrelated changes) unless --git-allow-dirty - always commits only if there are staged/unstaged changes """ self.stdout.write("[git] preparing auto-commit...") # Ensure we are in a git repo code, out, err = self._run_cmd(["git", "rev-parse", "--is-inside-work-tree"]) if code != 0 or out.lower() != "true": raise CommandError(f"[git] not a git repository (rev-parse failed): {err or out}") # Check status code, out, err = self._run_cmd(["git", "status", "--porcelain"]) if code != 0: raise CommandError(f"[git] status failed: {err or out}") dirty = bool(out.strip()) if dirty and not allow_dirty: raise CommandError( "[git] repository has uncommitted changes. " "Refusing to auto-commit without --git-allow-dirty." ) # Stage all code, out, err = self._run_cmd(["git", "add", "-A"]) if code != 0: raise CommandError(f"[git] add failed: {err or out}") # Don’t commit if nothing to commit code, out, err = self._run_cmd(["git", "diff", "--cached", "--quiet"]) # git diff --quiet returns 0 if no diff, 1 if there is a diff if code == 0: self.stdout.write("[git] nothing to commit (index clean).") return # Commit code, out, err = self._run_cmd(["git", "commit", "-m", message]) if code != 0: raise CommandError(f"[git] commit failed: {err or out}") self.stdout.write(self.style.SUCCESS(f"[git] committed: {message}")) def _db_dump(self, db_alias: str, dump_dir: str, dump_tool: str, extra_args: str) -> None: """ MySQL/MariaDB dump. Credentials should NOT be printed. Best practice: configure ~/.my.cnf or /etc/mysql/conf.d with credentials for non-interactive cron. """ self.stdout.write("[dump] preparing DB dump...") conn = connections[db_alias] cfg = settings.DATABASES.get(db_alias, {}) engine = (cfg.get("ENGINE") or "").lower() if "mysql" not in engine: raise CommandError(f"[dump] DB ENGINE does not look like MySQL/MariaDB: {cfg.get('ENGINE')}") name = cfg.get("NAME") user = cfg.get("USER") or "" host = cfg.get("HOST") or "" port = str(cfg.get("PORT") or "") if not name: raise CommandError("[dump] DATABASE NAME is empty in settings.DATABASES") dump_path = Path(dump_dir).expanduser().resolve() dump_path.mkdir(parents=True, exist_ok=True) ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") out_file = dump_path / f"{name}_{db_alias}_{ts}.sql" # Pick tool tool = dump_tool if tool == "auto": # prefer mariadb-dump if available, else mysqldump for candidate in ("mariadb-dump", "mysqldump"): code, _o, _e = self._run_cmd(["bash", "-lc", f"command -v {candidate} >/dev/null 2>&1"]) if code == 0: tool = candidate break if tool == "auto": tool = "mysqldump" # Build command cmd = [tool] if extra_args: cmd += extra_args.split() # Connection params (avoid password here) if host: cmd += ["-h", host] if port: cmd += ["-P", port] if user: cmd += ["-u", user] cmd += [name] self.stdout.write(f"[dump] tool={tool} output={out_file}") self.stdout.write("[dump] NOTE: for cron, store credentials in ~/.my.cnf (recommended).") # Run and redirect stdout to file with open(out_file, "w", encoding="utf-8") as f: p = subprocess.Popen(cmd, stdout=f, stderr=subprocess.PIPE, text=True) _out, err = p.communicate() if p.returncode != 0: raise CommandError(f"[dump] dump failed: {err.strip()}") self.stdout.write(self.style.SUCCESS(f"[dump] OK: {out_file}")) def add_arguments(self, parser): parser.add_argument("--database", default=DEFAULT_DB_ALIAS, help="DB alias (default: 'default').") parser.add_argument( "--application", action="append", default=[], help="Target app label. Repeatable. Example: --application blog --application shop", ) parser.add_argument( "--datamodel", action="append", default=[], help="Target model(s) as app_label.ModelName. Repeatable.", ) parser.add_argument("--force", action="store_true", help="Required. Without this flag the command refuses.") parser.add_argument("--noinput", action="store_true", help="Cron-friendly: do not prompt.") parser.add_argument( "--backup-dir", default="", help="Optional. If set, copies deleted migration files to this folder (timestamped).", ) parser.add_argument("--skip-precheck", action="store_true", help="Skip safety checks (NOT recommended).") parser.add_argument( "--progress", default="bar", choices=["bar", "dots", "none"], help="Progress style: bar|dots|none (default: bar). Use 'none' for clean logs.", ) parser.add_argument( "--show-tables", action="store_true", help="If set, prints DB tables for targeted apps (introspection).", ) # --- Optional Git commit --- parser.add_argument( "--git-commit", action="store_true", help="(Optional) Run 'git add -A' and commit before destructive actions.", ) parser.add_argument( "--git-message", default="chore: reset migrations (auto)", help="Commit message used with --git-commit.", ) parser.add_argument( "--git-allow-dirty", action="store_true", help="Allow commit even if repo has unrelated changes (default: refuse if dirty).", ) # --- Optional DB dump (MySQL/MariaDB) --- parser.add_argument( "--db-dump", action="store_true", help="(Optional) Run a MySQL/MariaDB dump before destructive actions.", ) parser.add_argument( "--dump-dir", default="/var/backups/ideo-lab", help="Directory where DB dumps will be stored.", ) parser.add_argument( "--dump-tool", default="auto", choices=["auto", "mysqldump", "mariadb-dump"], help="Dump tool to use (default: auto).", ) parser.add_argument( "--dump-extra-args", default="--single-transaction --routines --triggers --events", help="Extra args passed to dump tool as a string.", ) # ----------------------------- # Main # ----------------------------- def handle(self, *args, **opts): db_alias: str = opts["database"] app_labels: List[str] = opts["application"] or [] datamodels: List[str] = opts["datamodel"] or [] force: bool = opts["force"] noinput: bool = opts["noinput"] backup_dir: str = (opts["backup_dir"] or "").strip() skip_precheck: bool = bool(opts["skip_precheck"]) progress_mode: str = opts["progress"] show_tables: bool = bool(opts["show_tables"]) if not force: raise CommandError("Refusing to run without --force (this command is destructive).") target_apps = self._resolve_target_apps(app_labels, datamodels) all_apps_mode = not bool(target_apps) self._log_header(db_alias, target_apps) if not noinput: self.stdout.write(self.style.WARNING( "This will DELETE migration files and DELETE rows from django_migrations." )) confirm = input("Type YES to continue: ").strip() if confirm != "YES": self.stdout.write(self.style.NOTICE("Aborted.")) return # Progress uses stderr so stdout logs remain clean-ish (esp cron) prog = Progress( enabled=(progress_mode != "none"), mode=progress_mode, width=34, stream=getattr(self, "stderr", None), ) # 0) Precheck if not skip_precheck: self._precheck(db_alias=db_alias, target_apps=target_apps) # Optional: show impacted tables (best-effort) if show_tables: self._log_tables(db_alias=db_alias, target_apps=target_apps) # Optional: git commit BEFORE removing migrations if opts.get("git_commit"): self._git_commit( message=opts.get("git_message"), allow_dirty=bool(opts.get("git_allow_dirty")), noinput=noinput, ) # Optional: DB dump BEFORE touching django_migrations if opts.get("db_dump"): self._db_dump( db_alias=db_alias, dump_dir=opts.get("dump_dir"), dump_tool=opts.get("dump_tool"), extra_args=opts.get("dump_extra_args"), ) # Backup folder if requested backup_path = None if backup_dir: ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = Path(backup_dir).expanduser().resolve() / f"migration_cleanup_backup_{ts}" backup_path.mkdir(parents=True, exist_ok=True) self.stdout.write(f"[backup] {backup_path}") # Discover migrations dirs mig_dirs = self._iter_migrations_dirs(target_apps) self.stdout.write(f"[scan] migrations/ dirs: {len(mig_dirs)}") for d in mig_dirs: self.stdout.write(f" - {d}") # 1) Delete migration *.py except __init__.py files_to_delete = self._list_migration_pyfiles(mig_dirs) self.stdout.write(f"[plan] migration files to delete: {len(files_to_delete)}") prog.start(total=len(files_to_delete), label="Deleting migration files") deleted_files = self._delete_files(files_to_delete, backup_path=backup_path, prog=prog) prog.done() # 2) Delete __pycache__ under migrations/ pycache_dirs = self._list_pycache_dirs(mig_dirs) self.stdout.write(f"[plan] __pycache__ dirs to delete: {len(pycache_dirs)}") prog.start(total=len(pycache_dirs), label="Deleting __pycache__") deleted_caches = self._delete_dirs(pycache_dirs, prog=prog) prog.done() # 3) Clear django_migrations table deleted_rows = self._clear_django_migrations(db_alias=db_alias, target_apps=target_apps) self.stdout.write(self.style.WARNING(f"[db] django_migrations rows deleted: {deleted_rows}")) # 4) makemigrations self.stdout.write("[run] makemigrations ...") if target_apps: self.stdout.write(f"[run] apps: {', '.join(sorted(target_apps))}") call_command("makemigrations", *sorted(target_apps)) else: self.stdout.write("[run] apps: ALL") call_command("makemigrations") # 5) migrate --fake self.stdout.write("[run] migrate --fake ...") if target_apps: for app_label in sorted(target_apps): self.stdout.write(f"[run] fake-migrate app: {app_label}") call_command("migrate", app_label, database=db_alias, fake=True) else: call_command("migrate", database=db_alias, fake=True) # Summary self.stdout.write(self.style.SUCCESS("=== DONE ===")) self.stdout.write(f"[summary] apps: {'ALL' if all_apps_mode else ', '.join(sorted(target_apps))}") self.stdout.write(f"[summary] migration files deleted: {len(deleted_files)}") self.stdout.write(f"[summary] __pycache__ dirs deleted: {len(deleted_caches)}") self.stdout.write(f"[summary] django_migrations rows deleted: {deleted_rows}") # ----------------------------- # Logging helpers # ----------------------------- def _log_header(self, db_alias: str, target_apps: Set[str]) -> None: self.stdout.write(self.style.WARNING("=== MIGRATION CLEANUP (DESTRUCTIVE) ===")) self.stdout.write(f"[ctx] database: {db_alias}") self.stdout.write(f"[ctx] target apps: {', '.join(sorted(target_apps)) if target_apps else 'ALL'}") self.stdout.write(f"[ctx] project: {getattr(settings, 'PROJECT_NAME', 'django-project')}") # ----------------------------- # Resolve target apps # ----------------------------- def _resolve_target_apps(self, app_labels: List[str], datamodels: List[str]) -> Set[str]: resolved: Set[str] = set() for a in app_labels: a = (a or "").strip() if not a: continue try: apps.get_app_config(a) except Exception as e: raise CommandError(f"Unknown app label '{a}': {e}") resolved.add(a) for dm in datamodels: dm = (dm or "").strip() if not dm: continue if "." not in dm: raise CommandError(f"Invalid --datamodel '{dm}'. Use app_label.ModelName") app_label, model_name = dm.split(".", 1) app_label = app_label.strip() model_name = model_name.strip() try: model = apps.get_model(app_label, model_name) except Exception as e: raise CommandError(f"Invalid --datamodel '{dm}': {e}") resolved.add(model._meta.app_label) return resolved # ----------------------------- # Precheck # ----------------------------- def _precheck(self, db_alias: str, target_apps: Set[str]) -> None: self.stdout.write("[0] precheck: pending migrations + model changes...") connection = connections[db_alias] executor = MigrationExecutor(connection) targets = executor.loader.graph.leaf_nodes() plan = executor.migration_plan(targets) if target_apps: plan = [p for p in plan if p[0].app_label in target_apps] if plan: sample = ", ".join([f"{m.app_label}.{m.name}" for (m, backward) in plan[:10]]) raise CommandError( "Precheck failed: unapplied migrations exist. Apply them first.\n" f"Examples: {sample}" ) # Check for model changes (makemigrations --check) try: if target_apps: call_command("makemigrations", *sorted(target_apps), dry_run=True, check=True, verbosity=0) else: call_command("makemigrations", dry_run=True, check=True, verbosity=0) except SystemExit: raise CommandError( "Precheck failed: model changes detected (makemigrations --check would create migrations). " "Fix/commit state first or use --skip-precheck." ) self.stdout.write(self.style.SUCCESS("[0] precheck OK")) # ----------------------------- # Migrations dirs & listing # ----------------------------- def _iter_migrations_dirs(self, target_apps: Set[str]) -> List[Path]: dirs: List[Path] = [] for app_config in apps.get_app_configs(): if target_apps and app_config.label not in target_apps: continue app_path = Path(app_config.path) mig_dir = app_path / "migrations" if mig_dir.exists() and mig_dir.is_dir(): dirs.append(mig_dir) return dirs def _list_migration_pyfiles(self, mig_dirs: List[Path]) -> List[Path]: files: List[Path] = [] for mig_dir in mig_dirs: for pyfile in mig_dir.glob("*.py"): if pyfile.name == "__init__.py": continue files.append(pyfile) return sorted(files) def _list_pycache_dirs(self, mig_dirs: List[Path]) -> List[Path]: caches: List[Path] = [] for mig_dir in mig_dirs: for root, dirnames, _filenames in os.walk(mig_dir): if "__pycache__" in dirnames: caches.append(Path(root) / "__pycache__") # unique uniq = sorted({p.resolve() for p in caches}) return uniq # ----------------------------- # Delete operations with progress # ----------------------------- def _delete_files(self, files: List[Path], backup_path: Optional[Path], prog: Progress) -> List[Path]: deleted: List[Path] = [] for f in files: if backup_path: # keep relative structure: /migrations/ rel = f.relative_to(Path(f).parents[1]) # migrations/ might be too shallow sometimes # safer: use app root name + migrations app_root = f.parents[1].name # app folder name dest = backup_path / app_root / rel dest.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(f, dest) try: f.unlink() deleted.append(f) except Exception as e: raise CommandError(f"Failed to delete {f}: {e}") prog.step(1) return deleted def _delete_dirs(self, dirs: List[Path], prog: Progress) -> List[Path]: deleted: List[Path] = [] for d in dirs: try: shutil.rmtree(d) deleted.append(d) except Exception as e: raise CommandError(f"Failed to delete {d}: {e}") prog.step(1) return deleted # ----------------------------- # DB operations # ----------------------------- def _clear_django_migrations(self, db_alias: str, target_apps: Set[str]) -> int: self.stdout.write("[3] clearing django_migrations ...") connection = connections[db_alias] table = "django_migrations" deleted_rows = 0 with connection.cursor() as cursor, transaction.atomic(using=db_alias): if target_apps: placeholders = ", ".join(["%s"] * len(target_apps)) sql = f"DELETE FROM {table} WHERE app IN ({placeholders})" cursor.execute(sql, list(sorted(target_apps))) deleted_rows = cursor.rowcount if cursor.rowcount is not None else 0 self.stdout.write(f"[3] deleted apps rows: {', '.join(sorted(target_apps))}") else: cursor.execute(f"DELETE FROM {table}") deleted_rows = cursor.rowcount if cursor.rowcount is not None else 0 self.stdout.write("[3] deleted ALL rows") return deleted_rows def _log_tables(self, db_alias: str, target_apps: Set[str]) -> None: """ Best-effort: print DB tables that correspond to app models (db_table). """ self.stdout.write("[info] tables impacted (from models.db_table) ...") table_map: Dict[str, List[str]] = {} for model in apps.get_models(): app_label = model._meta.app_label if target_apps and app_label not in target_apps: continue table_map.setdefault(app_label, []).append(model._meta.db_table) for app_label in sorted(table_map.keys()): tables = sorted(set(table_map[app_label])) self.stdout.write(f" - {app_label}: {len(tables)} tables") # avoid ultra spam: print max 40 per app for t in tables[:40]: self.stdout.write(f" {t}") if len(tables) > 40: self.stdout.write(f" ... (+{len(tables) - 40} more)")