Function ensure_target_empty

scripts/migrate_sqlite_to_postgres.py:119–128

(connection: Connection, plan: Sequence[TablePlan])

def ensure_target_empty(connection: Connection, plan: Sequence[TablePlan]) -> None:
    non_empty: list[str] = []
    for item in plan:
        count = connection.execute(select(func.count()).select_from(item.target_table)).scalar_one()
        if count:
            non_empty.append(f"{item.name}={count}")
    if non_empty:
        raise MigrationError(
            "目标库不是空库。请加上 --truncate-target 后重试。非空表: " + ", ".join(non_empty[:8])
        )


def fetch_source_rows(connection: Connection, plan: TablePlan) -> list[dict]: