Prepare new meta field based on old/new tasks and old workflow task.
Source code in fractal_server/app/routes/api/v2/_aux_functions_task_version_update.py
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 | def get_new_workflow_task_meta(
*,
old_workflow_task_meta: dict | None,
old_task_meta: dict | None,
new_task_meta: dict | None,
) -> dict[str, Any] | None:
"""
Prepare new meta field based on old/new tasks and old workflow task.
"""
# When the whole `old_workflow_task_meta` is user-provided, use it
# as the outcome
if old_task_meta is None:
return old_workflow_task_meta
# When `old_workflow_task_meta` is unset, use the new-task meta as default.
if old_workflow_task_meta is None:
return new_task_meta
if new_task_meta is None:
new_task_meta = {}
# Find properties that were added to the old defaults
additions = {
k: v
for k, v in old_workflow_task_meta.items()
if v != old_task_meta.get(k)
}
# Find properties that were removed from the old defaults
removals = old_task_meta.keys() - old_workflow_task_meta.keys()
# Add `additions` and remove `removals`.
new_workflowtask_meta = {
k: v
for k, v in (new_task_meta | additions).items()
if k not in removals
}
return new_workflowtask_meta
|