This function creates the package manifest based on a task_list.py
Python module located in the dev subfolder of the package, see an
example of such list at ...
The manifest is then written to __FRACTAL_MANIFEST__.json, in the
main package directory.
Note: a valid example of custom_pydantic_models would be
[
("my_task_package", "some_module.py", "SomeModel"),
]
| PARAMETER |
DESCRIPTION |
package |
The name of the package (must be importable).
TYPE:
str
DEFAULT:
'fractal_tasks_core'
|
manifest_version |
TYPE:
str
DEFAULT:
'2'
|
has_args_schemas |
Whether to autogenerate JSON Schemas for task arguments.
TYPE:
bool
DEFAULT:
True
|
custom_pydantic_models |
Custom models to be included when building JSON Schemas for task
arguments.
TYPE:
Optional[list[tuple[str, str, str]]]
DEFAULT:
None
|
Source code in fractal_tasks_core/dev/create_manifest.py
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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147 | def create_manifest(
package: str = "fractal_tasks_core",
manifest_version: str = "2",
has_args_schemas: bool = True,
docs_link: Optional[str] = None,
custom_pydantic_models: Optional[list[tuple[str, str, str]]] = None,
):
"""
This function creates the package manifest based on a `task_list.py`
Python module located in the `dev` subfolder of the package, see an
example of such list at ...
The manifest is then written to `__FRACTAL_MANIFEST__.json`, in the
main `package` directory.
Note: a valid example of `custom_pydantic_models` would be
```
[
("my_task_package", "some_module.py", "SomeModel"),
]
```
Arguments:
package: The name of the package (must be importable).
manifest_version: Only `"2"` is supported.
has_args_schemas:
Whether to autogenerate JSON Schemas for task arguments.
custom_pydantic_models:
Custom models to be included when building JSON Schemas for task
arguments.
"""
# Preliminary check
if manifest_version != "2":
raise NotImplementedError(f"{manifest_version=} is not supported")
logging.info("Start generating a new manifest")
# Prepare an empty manifest
manifest = dict(
manifest_version=manifest_version,
task_list=[],
has_args_schemas=has_args_schemas,
)
if has_args_schemas:
manifest["args_schema_version"] = ARGS_SCHEMA_VERSION
# Prepare a default value of docs_link
if package == "fractal_tasks_core" and docs_link is None:
docs_link = (
"https://fractal-analytics-platform.github.io/fractal-tasks-core"
)
# Import the task list from `dev/task_list.py`
task_list_module = import_module(f"{package}.dev.task_list")
TASK_LIST = getattr(task_list_module, "TASK_LIST")
# Loop over TASK_LIST, and append the proper task dictionary
# to manifest["task_list"]
for task_obj in TASK_LIST:
# Convert Pydantic object to dictionary
task_dict = task_obj.model_dump(
exclude={"meta_init", "executable_init", "meta", "executable"},
exclude_unset=True,
)
# Copy some properties from `task_obj` to `task_dict`
if task_obj.executable_non_parallel is not None:
task_dict[
"executable_non_parallel"
] = task_obj.executable_non_parallel
if task_obj.executable_parallel is not None:
task_dict["executable_parallel"] = task_obj.executable_parallel
if task_obj.meta_non_parallel is not None:
task_dict["meta_non_parallel"] = task_obj.meta_non_parallel
if task_obj.meta_parallel is not None:
task_dict["meta_parallel"] = task_obj.meta_parallel
# Autogenerate JSON Schemas for non-parallel/parallel task arguments
if has_args_schemas:
for kind in ["non_parallel", "parallel"]:
executable = task_dict.get(f"executable_{kind}")
if executable is not None:
logging.info(f"[{executable}] START")
schema = create_schema_for_single_task(
executable,
package=package,
custom_pydantic_models=custom_pydantic_models,
)
logging.info(f"[{executable}] END (new schema)")
task_dict[f"args_schema_{kind}"] = schema
# Update docs_info, based on task-function description
docs_info = create_docs_info(
executable_non_parallel=task_obj.executable_non_parallel,
executable_parallel=task_obj.executable_parallel,
package=package,
)
if docs_info is not None:
task_dict["docs_info"] = docs_info
if docs_link is not None:
task_dict["docs_link"] = docs_link
manifest["task_list"].append(task_dict)
print()
# Write manifest
imported_package = import_module(package)
manifest_path = (
Path(imported_package.__file__).parent / "__FRACTAL_MANIFEST__.json"
)
with manifest_path.open("w") as f:
json.dump(manifest, f, indent=2)
f.write("\n")
logging.info(f"Manifest stored in {manifest_path.as_posix()}")
|