Files
Alexei Fedotov 5852ae55b0 Praktika: fix workflow_call generation for booleans, per-job secrets, empty inputs
Address review feedback on the reusable-workflow generator:

- Thread is_required through TEMPLATE_BOOLEAN_INPUT so a required boolean
  input keeps its 'must choose' contract instead of silently defaulting to
  false in both workflow_dispatch and workflow_call.
- Re-declare per-job GH secrets (Job.Config.secrets) under
  on.workflow_call.secrets in addition to workflow-level secrets,
  deduplicated and order-preserving, so callers can pass every secret the
  generated jobs reference.
- Suppress the whole 'inputs:' stanza for input-less dispatch workflows so
  they no longer emit empty workflow_dispatch.inputs / workflow_call.inputs
  blocks.
2026-07-07 13:51:05 +02:00

663 lines
24 KiB
Python

import dataclasses
import math
from typing import List
from . import Artifact, Job, Workflow
from .mangle import _get_workflows
from .parser import WorkflowConfigParser
from .settings import Settings
from .utils import Shell, Utils
class YamlGenerator:
class Templates:
TEMPLATE_MERGE_QUEUE_0 = """\
# generated by praktika
name: {NAME}
on:
merge_group:
env:
# Force the stdout and stderr streams to be unbuffered
PYTHONUNBUFFERED: 1
{ENV_CHECKOUT_REFERENCE}
{ENV_SECRETS}
jobs:
{JOBS}\
"""
TEMPLATE_PULL_REQUEST_0 = """\
# generated by praktika
name: {NAME}
on:
{EVENT}:
branches: [{BRANCHES}]
env:
# Force the stdout and stderr streams to be unbuffered
PYTHONUNBUFFERED: 1
{ENV_CHECKOUT_REFERENCE}
{ENV_SECRETS}
{GH_TOKEN_PERMISSIONS}
jobs:
{JOBS}\
"""
TEMPLATE_GH_TOKEN_PERMISSIONS = """\
# Allow updating GH commit statuses and PR comments to post an actual job reports link
permissions: write-all\
"""
TEMPLATE_ENV_CHECKOUT_REF_PR = """\
DISABLE_CI_MERGE_COMMIT: ${{{{ vars.DISABLE_CI_MERGE_COMMIT || '0' }}}}
DISABLE_CI_CACHE: ${{{{ vars.DISABLE_CI_CACHE || '0' }}}}
CHECKOUT_REF: ${{{{ vars.DISABLE_CI_MERGE_COMMIT == '1' && github.event.pull_request.head.sha || '' }}}}\
"""
TEMPLATE_ENV_CHECKOUT_REF_DEFAULT = """\
CHECKOUT_REF: ""\
"""
TEMPLATE_ENV_SECRET = """\
{SECRET_NAME}: ${{{{{{{{ secrets.{SECRET_NAME} }}}}}}}}
"""
TEMPLATE_ENV_GH_VAR = """\
{VAR_NAME}: ${{{{{{{{ vars.{VAR_NAME} }}}}}}}}
"""
TEMPLATE_SCHEDULE = """\
# generated by praktika
name: {NAME}
on:
schedule:{CRON_TEMPLATES}
workflow_dispatch:{DISPATCH_INPUTS_BLOCK}
concurrency:
group: ${{{{{{{{ github.workflow }}}}}}}}
env:
PYTHONUNBUFFERED: 1
{ENV_CHECKOUT_REFERENCE}
{ENV_SECRETS}
jobs:
{JOBS}\
"""
TEMPLATE_CRON = """
- cron: {CRON_SCHEDULE}\
"""
TEMPLATE_DISPATCH_WORKFLOW = """\
# generated by praktika
name: {NAME}
concurrency:
group: ${{{{{{{{ github.workflow }}}}}}}}
on:
workflow_dispatch:{DISPATCH_INPUTS_BLOCK}{WORKFLOW_CALL}
env:
PYTHONUNBUFFERED: 1
{ENV_CHECKOUT_REFERENCE}
{ENV_SECRETS}
{GH_TOKEN_PERMISSIONS}
jobs:
{JOBS}\
"""
TEMPLATE_WORKFLOW_CALL = """
workflow_call:{INPUTS_BLOCK}\
"""
TEMPLATE_WORKFLOW_CALL_STRING_INPUT = """
{NAME}:
description: {DESCRIPTION}
required: {IS_REQUIRED}
type: string
default: {DEFAULT_VALUE}\
"""
TEMPLATE_INPUT = """
{NAME}:
description: {DESCRIPTION}
required: {IS_REQUIRED}
default: {DEFAULT_VALUE}\
"""
TEMPLATE_OPTIONS_INPUT = """
{NAME}:
description: {DESCRIPTION}
type: choice
options: {OPTIONS}
default: {DEFAULT_VALUE}\
"""
TEMPLATE_BOOLEAN_INPUT = """
{NAME}:
description: {DESCRIPTION}
required: {IS_REQUIRED}
type: boolean
default: {DEFAULT_VALUE}\
"""
TEMPLATE_SECRET_CONFIG = """\
{SECRET_NAME}:
required: true
"""
TEMPLATE_MATRIX = """
strategy:
fail-fast: false
matrix:
params: {PARAMS_LIST}\
"""
TEMPLATE_JOB_0 = """
{JOB_NAME_NORMALIZED}:
runs-on: [{RUNS_ON}]
needs: [{NEEDS}]{IF_EXPRESSION}
name: "{JOB_NAME_GH}"{TIMEOUT_MINUTES}
outputs:
data: ${{{{ steps.run.outputs.DATA }}}}
pipeline_status: ${{{{ steps.run.outputs.pipeline_status || 'undefined' }}}}
steps:
- name: Checkout code
uses: actions/checkout@v6
with:
ref: ${{{{ env.CHECKOUT_REF }}}}
{JOB_ADDONS}
- name: Prepare env script
run: |
rm -rf {UNIQUE_WORK_DIRS}
mkdir -p {UNIQUE_WORK_DIRS}
cat > {ENV_SETUP_SCRIPT} << 'ENV_SETUP_SCRIPT_EOF'
export PYTHONPATH=./ci:.:{PYTHONPATH_EXTRA}
{SETUP_ENVS}
cat > {WORKFLOW_JOB_FILE} << 'EOF'
${{{{ toJson(job) }}}}
EOF
cat > {WORKFLOW_STATUS_FILE} << 'EOF'
${{{{ toJson(needs) }}}}
EOF
ENV_SETUP_SCRIPT_EOF
{DOWNLOADS_GITHUB}
- name: Run
id: run
run: |
. {ENV_SETUP_SCRIPT}
PYTHONUNBUFFERED=1 python3 -m praktika run '{JOB_NAME}' --workflow "{WORKFLOW_NAME}" --ci --timestamp
{UPLOADS_GITHUB}\
"""
TEMPLATE_SETUP_ENV_SECRETS = """\
export {SECRET_NAME}=$(cat<<'EOF'
${{{{ secrets.{SECRET_NAME} }}}}
EOF
)\
"""
TEMPLATE_SETUP_ENV_VARS = """\
export {VAR_NAME}=$(cat<<'EOF'
${{{{ vars.{VAR_NAME} }}}}
EOF
)\
"""
TEMPLATE_SETUP_ENVS_INPUTS = """\
cat > {WORKFLOW_INPUTS_FILE} << 'EOF'
{INPUTS_EXPR}
EOF\
"""
# The `inputs` context holds this workflow's own inputs for both
# workflow_dispatch and workflow_call, so it is the correct source when
# workflow_call is enabled. (Testing `github.event.inputs` truthiness is
# wrong: when the reusable workflow is called from a manually-dispatched
# caller, `github.event.inputs` is the *caller's* inputs, which would
# drop this workflow's `ref`/`type`.) For dispatch-only workflows we keep
# `github.event.inputs` for backward compatibility.
WORKFLOW_INPUTS_EXPR_DISPATCH = "${{ toJson(github.event.inputs) }}"
WORKFLOW_INPUTS_EXPR_WITH_CALL = "${{ toJson(inputs) }}"
TEMPLATE_PY_INSTALL = """
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: {PYTHON_VERSION}
"""
TEMPLATE_PY_WITH_REQUIREMENTS = """
- name: Install dependencies
run: |
sudo apt-get update && sudo apt install -y python3-pip
# TODO: --break-system-packages? otherwise ubuntu's apt/apt-get complains
{PYTHON} -m pip install --upgrade pip --break-system-packages
{PIP} install -r {REQUIREMENT_PATH} --break-system-packages
"""
TEMPLATE_GH_UPLOAD = """
- name: Upload artifact {NAME}
uses: actions/upload-artifact@v4
with:
name: {NAME}
path: {PATH}
"""
TEMPLATE_GH_DOWNLOAD = """
- name: Download artifact {NAME}
uses: actions/download-artifact@v4
with:
name: {NAME}
path: {PATH}
"""
# Per-job run condition emitted for every cacheable job. The job runs
# only when:
# * the workflow was not cancelled, and
# * no upstream dependency reported 'failure' or 'undefined', and
# * the config_workflow job did NOT mark this job as a cache hit.
# `cache_success_base64` is the list of base64-encoded job names that
# the config_workflow resolved to a previous successful run (see
# `hook_cache.py`); `{JOB_NAME_BASE64}` is this job's name base64'd.
# When the name is present, a cached success exists and the job is
# skipped. NOTE: this token is the job NAME, so renaming a job changes
# the token and invalidates its old cache entries (the stale entries
# are simply left behind and ignored).
TEMPLATE_IF_EXPRESSION = """
if: ${{{{ !cancelled() && !contains(needs.*.outputs.pipeline_status, 'failure') && !contains(needs.*.outputs.pipeline_status, 'undefined') && !contains(fromJson(needs.{WORKFLOW_CONFIG_JOB_NAME}.outputs.data).workflow_config.cache_success_base64, '{JOB_NAME_BASE64}') }}}}\
"""
TEMPLATE_IF_EXPRESSION_NOT_CANCELLED = """
if: ${{ !cancelled() }}\
"""
TEMPLATE_IF_EXPRESSION_ALWAYS = """
if: ${{ always() }}\
"""
def __init__(self):
self.py_workflows = [] # type: List[Workflow.Config]
@classmethod
def _get_workflow_file_name(cls, file_name):
yaml_name = file_name.removesuffix(".py") + ".yml"
return f"{Settings.WORKFLOW_PATH_PREFIX}/{Utils.normalize_string(yaml_name)}"
def generate(self):
print("---Start generating yaml pipelines---")
files = []
self.py_workflows = _get_workflows(_file_names_out=files)
assert self.py_workflows and files
for workflow_config, workflow_file_name in zip(self.py_workflows, files):
print(f"Generate workflow [{workflow_config.name}]")
parser = WorkflowConfigParser(workflow_config).parse()
yaml_workflow_str = PullRequestPushYamlGen(parser).generate()
with open(self._get_workflow_file_name(workflow_file_name), "w") as f:
f.write(yaml_workflow_str)
Shell.check("git add ./.github/workflows/*.yml")
class PullRequestPushYamlGen:
def __init__(self, parser: WorkflowConfigParser):
self.workflow_config = parser.workflow_yaml_config
self.parser = parser
def generate(self):
# Propagate transitive dependencies so that GH Actions expressions like
# `!contains(needs.*.outputs.pipeline_status, 'failure')` see the full
# upstream chain. Example: A -> B -> C. If A fails then B is skipped;
# without transitive `needs`, C may not see A in `needs.*`.
_memo: dict = {}
def _all_needs(job_name: str) -> set:
if job_name in _memo:
return _memo[job_name]
_memo[job_name] = set() # guard against cycles
result = set(self.workflow_config.job_to_config[job_name].needs)
for dep in list(result):
result |= _all_needs(dep)
_memo[job_name] = result
return result
job_items = []
for i, job in enumerate(self.workflow_config.jobs):
job_name_normalized = Utils.normalize_string(job.name)
needs = ", ".join(sorted(map(Utils.normalize_string, _all_needs(job.name))))
job_name = job.name
job_addons = []
for addon in job.addons:
if addon.install_python:
job_addons.append(
YamlGenerator.Templates.TEMPLATE_PY_INSTALL.format(
PYTHON_VERSION=Settings.PYTHON_VERSION
)
)
if addon.requirements_txt_path:
job_addons.append(
YamlGenerator.Templates.TEMPLATE_PY_WITH_REQUIREMENTS.format(
PYTHON=Settings.PYTHON_INTERPRETER,
PIP=Settings.PYTHON_PACKET_MANAGER,
PYTHON_VERSION=Settings.PYTHON_VERSION,
REQUIREMENT_PATH=addon.requirements_txt_path,
)
)
uploads_github = []
for artifact in job.artifacts_gh_provides:
uploads_github.append(
YamlGenerator.Templates.TEMPLATE_GH_UPLOAD.format(
NAME=artifact.name, PATH=artifact.path
)
)
downloads_github = []
for artifact in job.artifacts_gh_requires:
downloads_github.append(
YamlGenerator.Templates.TEMPLATE_GH_DOWNLOAD.format(
NAME=artifact.name, PATH=Settings.INPUT_DIR
)
)
config_job_name_normalized = Utils.normalize_string(
Settings.CI_CONFIG_JOB_NAME
)
if_expression = ""
if (
self.workflow_config.config.enable_cache
and job_name_normalized != config_job_name_normalized
):
if_expression = YamlGenerator.Templates.TEMPLATE_IF_EXPRESSION.format(
WORKFLOW_CONFIG_JOB_NAME=config_job_name_normalized,
JOB_NAME_BASE64=Utils.to_base64(job_name),
)
if job.run_unless_cancelled:
if_expression = (
YamlGenerator.Templates.TEMPLATE_IF_EXPRESSION_NOT_CANCELLED
)
if job.name == Settings.FINISH_WORKFLOW_JOB_NAME:
if_expression = YamlGenerator.Templates.TEMPLATE_IF_EXPRESSION_ALWAYS
# Emit timeout-minutes for any job whose configured timeout exceeds GitHub's 6h default.
# JobYaml has no timeout; get it from the original Job.Config in workflow config.
timeout_minutes = ""
orig_job = next(
(j for j in self.workflow_config.config.jobs if j.name == job.name),
None,
)
if (
orig_job
and getattr(orig_job, "timeout", None)
and orig_job.timeout > 360 * 60
):
timeout_minutes = (
f"\n timeout-minutes: {math.ceil(orig_job.timeout / 60) + 5}"
)
secrets_envs = []
for secret in job.secret_names_gh:
secrets_envs.append(
YamlGenerator.Templates.TEMPLATE_SETUP_ENV_SECRETS.format(
SECRET_NAME=secret
)
)
for var in job.variable_names_gh:
secrets_envs.append(
YamlGenerator.Templates.TEMPLATE_SETUP_ENV_VARS.format(VAR_NAME=var)
)
if (
self.workflow_config.event == Workflow.Event.DISPATCH
or self.workflow_config.dispatch_inputs
):
# Dispatch workflows always emit a `workflow_call` trigger, so
# they must read inputs from the `inputs` context too.
inputs_expr = (
YamlGenerator.Templates.WORKFLOW_INPUTS_EXPR_WITH_CALL
if self.workflow_config.event == Workflow.Event.DISPATCH
else YamlGenerator.Templates.WORKFLOW_INPUTS_EXPR_DISPATCH
)
secrets_envs.append(
YamlGenerator.Templates.TEMPLATE_SETUP_ENVS_INPUTS.format(
WORKFLOW_INPUTS_FILE=Settings.WORKFLOW_INPUTS_FILE,
INPUTS_EXPR=inputs_expr,
)
)
job_item = YamlGenerator.Templates.TEMPLATE_JOB_0.format(
JOB_NAME_NORMALIZED=job_name_normalized,
IF_EXPRESSION=if_expression,
TIMEOUT_MINUTES=timeout_minutes,
RUNS_ON=", ".join(job.runs_on),
NEEDS=needs,
JOB_NAME_GH=job_name.replace('"', '\\"'),
JOB_NAME=job_name.replace(
"'", "'\\''"
), # ' must be escaped so that yaml commands are properly parsed
WORKFLOW_NAME=self.workflow_config.name,
ENV_SETUP_SCRIPT=Settings.ENV_SETUP_SCRIPT,
SETUP_ENVS="\n".join(secrets_envs),
JOB_ADDONS="".join(job_addons),
DOWNLOADS_GITHUB="\n".join(downloads_github),
UPLOADS_GITHUB="\n".join(uploads_github),
PYTHON=Settings.PYTHON_INTERPRETER,
WORKFLOW_JOB_FILE=Settings.WORKFLOW_JOB_FILE,
WORKFLOW_STATUS_FILE=Settings.WORKFLOW_STATUS_FILE,
TEMP_DIR=Settings.TEMP_DIR,
UNIQUE_WORK_DIRS=" ".join(
{Settings.TEMP_DIR, Settings.INPUT_DIR, Settings.OUTPUT_DIR}
),
PYTHONPATH_EXTRA=Settings.PYTHONPATHS,
)
job_items.append(job_item)
# for schedule workflows only
cron_items = ""
for cron_item in self.workflow_config.cron_schedules:
cron_items += YamlGenerator.Templates.TEMPLATE_CRON.format(
CRON_SCHEDULE=cron_item
)
# for dispatch workflows only
dispatch_inputs = ""
for input_item in self.workflow_config.dispatch_inputs:
if input_item.is_boolean:
dispatch_inputs += YamlGenerator.Templates.TEMPLATE_BOOLEAN_INPUT.format(
NAME=input_item.name,
DESCRIPTION=input_item.description,
IS_REQUIRED="true" if input_item.is_required else "false",
DEFAULT_VALUE=input_item.default_value or "false",
)
elif not input_item.options:
dispatch_inputs += YamlGenerator.Templates.TEMPLATE_INPUT.format(
NAME=input_item.name,
DESCRIPTION=input_item.description,
IS_REQUIRED="true" if input_item.is_required else "false",
DEFAULT_VALUE=input_item.default_value or "''",
)
else:
dispatch_inputs += (
YamlGenerator.Templates.TEMPLATE_OPTIONS_INPUT.format(
NAME=input_item.name,
DESCRIPTION=input_item.description,
OPTIONS=input_item.options,
DEFAULT_VALUE=input_item.default_value or "''",
)
)
if self.workflow_config.event in (
Workflow.Event.PULL_REQUEST,
Workflow.Event.PUSH,
):
base_template = YamlGenerator.Templates.TEMPLATE_PULL_REQUEST_0
format_kwargs = {
"BRANCHES": ", ".join(
[f"'{branch}'" for branch in self.workflow_config.branches]
),
"EVENT": self.workflow_config.event,
"GH_TOKEN_PERMISSIONS": (
YamlGenerator.Templates.TEMPLATE_GH_TOKEN_PERMISSIONS
# if not Settings.USE_CUSTOM_GH_AUTH
# else ""
),
}
if self.workflow_config.event in (Workflow.Event.PULL_REQUEST,):
ENV_CHECKOUT_REFERENCE = (
YamlGenerator.Templates.TEMPLATE_ENV_CHECKOUT_REF_PR
)
else:
ENV_CHECKOUT_REFERENCE = (
YamlGenerator.Templates.TEMPLATE_ENV_CHECKOUT_REF_DEFAULT
)
elif self.workflow_config.event in (Workflow.Event.SCHEDULE,):
base_template = YamlGenerator.Templates.TEMPLATE_SCHEDULE
format_kwargs = {
"CRON_TEMPLATES": cron_items,
# Allow a scheduled workflow to also declare workflow_dispatch
# inputs for manual runs; empty when none are declared, so the
# generated YAML is unchanged for inputs-less schedules.
"DISPATCH_INPUTS_BLOCK": (
f"\n inputs:{dispatch_inputs}" if dispatch_inputs else ""
),
}
ENV_CHECKOUT_REFERENCE = (
YamlGenerator.Templates.TEMPLATE_ENV_CHECKOUT_REF_DEFAULT
)
elif self.workflow_config.event in (Workflow.Event.DISPATCH,):
base_template = YamlGenerator.Templates.TEMPLATE_DISPATCH_WORKFLOW
# Always emit a `workflow_call` trigger for dispatch workflows so any
# of them can be reused via `uses:` (e.g. auto_releases.yml calls
# CreateRelease), keeping dispatch and workflow_call consistent.
wc_inputs = ""
for input_item in self.workflow_config.dispatch_inputs:
if input_item.is_boolean:
wc_inputs += YamlGenerator.Templates.TEMPLATE_BOOLEAN_INPUT.format(
NAME=input_item.name,
DESCRIPTION=input_item.description,
IS_REQUIRED="true" if input_item.is_required else "false",
DEFAULT_VALUE=input_item.default_value or "false",
)
else:
# `workflow_call` has no `choice` type — expose as string.
wc_inputs += YamlGenerator.Templates.TEMPLATE_WORKFLOW_CALL_STRING_INPUT.format(
NAME=input_item.name,
DESCRIPTION=input_item.description,
IS_REQUIRED="true" if input_item.is_required else "false",
DEFAULT_VALUE=input_item.default_value or "''",
)
workflow_call = YamlGenerator.Templates.TEMPLATE_WORKFLOW_CALL.format(
INPUTS_BLOCK=f"\n inputs:{wc_inputs}" if wc_inputs else ""
)
# Reusable-workflow callers can only pass secrets declared under
# `on.workflow_call.secrets`, so re-declare every GH secret the
# generated jobs may reference — both workflow-level secrets and
# per-job secrets — deduplicated and order-preserving.
wc_secret_names = []
for secret in self.parser.config.secrets:
if secret.is_gh_secret() and secret.name not in wc_secret_names:
wc_secret_names.append(secret.name)
for job in self.parser.config.jobs:
for secret in job.secrets:
if secret.is_gh_secret() and secret.name not in wc_secret_names:
wc_secret_names.append(secret.name)
if wc_secret_names:
workflow_call += "\n secrets:"
for name in wc_secret_names:
workflow_call += f"\n {name}:\n required: true"
format_kwargs = {
"DISPATCH_INPUTS_BLOCK": (
f"\n inputs:{dispatch_inputs}" if dispatch_inputs else ""
),
"WORKFLOW_CALL": workflow_call,
"GH_TOKEN_PERMISSIONS": (
YamlGenerator.Templates.TEMPLATE_GH_TOKEN_PERMISSIONS
),
}
ENV_CHECKOUT_REFERENCE = (
YamlGenerator.Templates.TEMPLATE_ENV_CHECKOUT_REF_DEFAULT
)
elif self.workflow_config.event in (Workflow.Event.MERGE_QUEUE,):
base_template = YamlGenerator.Templates.TEMPLATE_MERGE_QUEUE_0
format_kwargs = {}
ENV_CHECKOUT_REFERENCE = (
YamlGenerator.Templates.TEMPLATE_ENV_CHECKOUT_REF_DEFAULT
)
else:
assert (
False
), f"Invalid or Not implemented event [{self.workflow_config.event}]"
SECRET_ENVS = ""
GH_VAR_ENVS = ""
for secret in self.parser.config.secrets:
if secret.is_gh_secret():
SECRET_ENVS += YamlGenerator.Templates.TEMPLATE_ENV_SECRET.format(
SECRET_NAME=secret.name
)
elif secret.is_gh_var():
GH_VAR_ENVS += YamlGenerator.Templates.TEMPLATE_ENV_GH_VAR.format(
VAR_NAME=secret.name
)
format_kwargs["ENV_SECRETS"] = GH_VAR_ENVS + SECRET_ENVS
template_1 = base_template.strip().format(
NAME=self.workflow_config.name,
JOBS="{}" * len(job_items),
ENV_CHECKOUT_REFERENCE=ENV_CHECKOUT_REFERENCE,
**format_kwargs,
)
res = template_1.format(*job_items)
return res
@dataclasses.dataclass
class AuxConfig:
# defines aux step to install dependencies
addon: Job.Requirements
# defines aux step(s) to upload GH artifacts
uploads_gh: List[Artifact.Config]
# defines aux step(s) to download GH artifacts
downloads_gh: List[Artifact.Config]
def get_aux_workflow_name(self):
suffix = ""
if self.addon.python_requirements_txt:
suffix += "_py"
for _ in self.uploads_gh:
suffix += "_uplgh"
for _ in self.downloads_gh:
suffix += "_dnlgh"
return f"{Settings.WORKFLOW_PATH_PREFIX}/aux_job{suffix}.yml"
def get_aux_workflow_input(self):
res = ""
if self.addon.python_requirements_txt:
res += f" requirements_txt: {self.addon.python_requirements_txt}"
return res
if __name__ == "__main__":
WFS = [
Workflow.Config(
name="PR",
event=Workflow.Event.PULL_REQUEST,
jobs=[
Job.Config(
name="Hello World",
runs_on=["foo"],
command="bar",
job_requirements=Job.Requirements(
python_requirements_txt="./requirement.txt"
),
)
],
enable_cache=True,
)
]
YamlGenerator().generate(workflow_config=WFS)