Rozdział 23 · Część IV · Wdrażamy SafeSort

Bezpiecznie przenoszcie pliki

Bezpośrednie sortowanie jest wykonywane przez apply_plan(): przed każdym ruchem, funkcja ponownie sprawdza konflikt. Undo pozostaje osobną operacją odwrotną.

Issue #5 · Project SafeSort Pierwsze wydanie

Add explicit apply operation

Area: SafetyPriority: High

git switch -c feat/apply-operation

Do tej pory żadna funkcja w łańcuchu bezpośredniego sortowania nie zmieniła SafeSort pliku system. Moduł executor.py wykonuje operację bezpośrednią: Funkcja apply_plan() ma gotowy plan i tylko się porusza pliki są w nim wymienione. Odwracaj działanie undo też przenosi pliki, ale przywraca je zgodnie z zarejestrowanym manifestem.

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

Zwracaj uwagę na weryfikację destination.exists() prosty Przed przeprowadzką, choć planista już uniknął tego konfliktu podczas fazy realizacji planu. Między tworzeniem planu a jego realizacją na dysku mógł pojawić się nowy plik, na przykład, jeśli sam użytkownik w danym momencie coś tam umieścił. Brak ponownej weryfikacji shutil.move() na systemach opartych na POSIX bezszelestnie nadpisywały Taki plik jest SafeSort nie nadpisuje bezgłośnie plików pod żadnym pozorem.

punkt kontrolny testu: apply zmienia tylko podaną ścieżkę

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()