- Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathexecution.py
More file actions
Latest commit
61 lines (55 loc) · 2.31 KB
/
Copy pathexecution.py
File metadata and controls
61 lines (55 loc) · 2.31 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
"""Filesystem execution for a reviewed, validated rename plan."""
from __future__ importannotations
importos
fromcollections.abcimportCallable
fromrename_planimportPlanIssue, RenameOperation, RenamePlan, plan_digest, validate_plan
ProgressCallback=Callable[[RenameOperation, str, str|None], None]
defapply_plan(
plan: RenamePlan,
rename_log_path: str|os.PathLike[str] |None=None,
progress: ProgressCallback|None=None,
) ->tuple[PlanIssue, ...]:
"""Revalidate, apply no-replace moves, and checkpoint progress when requested."""
ifplan_digest(plan.operations) !=plan.digest:
raiseValueError("rename-plan changed after validation")
issues=validate_plan(plan)
ifissues:
returnissues
completed: list[RenameOperation] = []
foroperationinplan.operations:
ifoperation.source==operation.destination:
continue
try:
os.link(operation.source, operation.destination)
os.unlink(operation.source)
completed.append(operation)
ifprogress:
progress(operation, "applied", None)
exceptOSErroraserror:
ifprogress:
progress(operation, "failed", str(error))
forcompleted_operationinreversed(completed):
try:
os.link(completed_operation.destination, completed_operation.source)
os.unlink(completed_operation.destination)
ifprogress:
progress(completed_operation, "rolled_back", None)
exceptOSErrorasrollback_error:
ifprogress:
progress(completed_operation, "rollback_failed", str(rollback_error))
return (
PlanIssue(
operation.scene_id,
operation.source,
operation.destination,
"apply_failed",
str(error),
),
)
ifrename_log_pathandcompleted:
withopen(rename_log_path, "a", encoding="utf-8") asrename_log:
foroperationincompleted:
rename_log.write(
"{}|{}|{}\n".format(operation.scene_id, operation.source, operation.destination)
)
return ()