Глава 23 · Часть IV · Реализуем SafeSort

Безопасно перемещаем файлы

Прямую сортировку выполняет apply_plan(): перед каждым перемещением функция повторно проверяет конфликт. Undo остаётся отдельной обратной операцией.

Issue #5 · Project «SafeSort — первый релиз»

Add explicit apply operation

Area: SafetyPriority: High

git switch -c feat/apply-operation

До сих пор ни одна функция в прямой цепочке сортировки SafeSort не меняла файловую систему. Модуль executor.py выполняет прямую операцию: функция apply_plan() получает готовый план и перемещает только перечисленные в нём файлы. Обратная операция undo тоже перемещает файлы, но восстанавливает их по записанному манифесту.

def apply_plan(plan: SortPlan) -> list[CompletedMove]:
    results = []
    for operation in plan.operations:
        source, destination = operation.source, operation.destination
        destination.parent.mkdir(parents=True, exist_ok=True)

        if destination.exists() or destination.is_symlink():
            results.append(CompletedMove(source, destination, completed=False,
                                          error="destination already exists"))
            continue

        shutil.move(str(source), str(destination))
        results.append(CompletedMove(source, destination, completed=True))
    return results

Обратите внимание на проверку destination.exists() прямо перед перемещением, хотя планировщик уже избегал этого конфликта на этапе построения плана. Между построением плана и его выполнением на диске мог появиться новый файл — например, если пользователь сам что-то туда положил в этот момент. Без повторной проверки shutil.move() в системах на основе POSIX молча перезаписал бы такой файл — а SafeSort не перезаписывает файлы молча ни при каких обстоятельствах.

Тест-чекпойнт: apply меняет только заявленный путь

def test_apply_plan_moves_one_file(tmp_path):
    source = tmp_path / "report.pdf"
    destination = tmp_path / "Sorted/documents/report.pdf"
    source.write_bytes(b"report")
    plan = SortPlan(tmp_path, (MoveOperation(source, destination),))

    [result] = apply_plan(plan)

    assert result.completed is True
    assert destination.read_bytes() == b"report"
    assert not source.exists()