Source code for dpgen2.entrypoint.workflow

import argparse
import json
import logging
import os
from typing import (
    Optional,
)

from dflow import (
    Workflow,
)

from dpgen2.entrypoint.args import normalize as normalize_args
from dpgen2.entrypoint.common import (
    global_config_workflow,
)

workflow_subcommands = [
    "terminate",
    "stop",
    "suspend",
    "delete",
    "retry",
    "resume",
]


[docs] def add_subparser_workflow_subcommand(subparsers, command: str): parser_cmd = subparsers.add_parser( command, help=f"{command.capitalize()} a DPGEN2 workflow.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser_cmd.add_argument("CONFIG", help="the config file in json format.") parser_cmd.add_argument("ID", help="the ID of the workflow.")
[docs] def execute_workflow_subcommand( command: str, wfid: str, wf_config: Optional[dict] = {}, ): wf_config = normalize_args(wf_config) global_config_workflow(wf_config) wf = Workflow(id=wfid) getattr(wf, command)()