Class MigrationAutodetector

tortoise/migrations/autodetector.py:37–199

class MigrationAutodetector:
    def __init__(
        self,
        apps: Apps,
        apps_config: dict[str, dict],
        *,
        now: Callable[[], dt.datetime] | None = None,
    ) -> None:
        self.apps = apps
        self.apps_config = apps_config
        self._now = now or dt.datetime.now
        self.loader = MigrationLoader(apps_config, _NoopRecorder(), load=False)

    async def changes(self) -> list[MigrationWriter]:
        await self.loader.build_graph()
        old_state = await self._project_state()
        new_state = self._current_state()
        writers: list[MigrationWriter] = []
        for app_label, config in self.apps_config.items():
            migrations_module = config.get("migrations")
            if not migrations_module:
                continue
            operations = OperationGenerator(old_state, new_state).generate(app_labels=[app_label])
            if not operations:
                continue
            dependencies = self._dependencies_for_app(app_label, new_state)
            name, initial = self._migration_name(app_label, old_state, new_state)
            writers.append(
                MigrationWriter(
                    name,
                    app_label,
                    operations,
                    dependencies=dependencies,
                    initial=initial,
                    migrations_module=migrations_module,
                )
            )
        return writers

    async def write(self) -> list[str]:
        writers = await self.changes()
        return [str(writer.write()) for writer in writers]

    def _current_state(self) -> State:
        state = State(models={}, apps=StateApps())
        for app_label, models in self.apps.items():
            for model in models.values():
                state.models[(app_label, model.__name__)] = ModelState.make_from_model(
                    app_label, model
                )
        return state

    async def _project_state(self) -> State:
        state = State(models={}, apps=StateApps())
        for key in self._full_plan():
            migration = self.loader.graph.nodes[key]
            if not isinstance(migration, Migration):
                raise ValueError(f"Missing migration for {key}")