workflows#

class cloudly.gcp.workflows.Workflow[source]#

Bases: object

classmethod create(name: str, config: WorkflowConfig, *, region: str) → Workflow[source]#

name needs to be unique, hence it’s recommended to construct it with some randomness.

If you create a workflow for a, say, batch job, then you probably should get region from the batch job definition. I don’t know whether it’s allowed for a workflow to contain jobs spanning regions.

classmethod list(region: str) → list[Workflow][source]#

__init__(name_or_obj: str | Workflow, /)[source]#: name is like “projects/<project_id>/locations/<region>/workflows/<name>”.

property region: str#

property definition: dict#

property create_time: datetime#

property update_time: datetime#

property revision_id: str#

state() → Literal['STATE_UNSPECIFIED', 'ACTIVE', 'UNAVAILABLE'][source]#

update(config: WorkflowConfig)[source]#

execute(args: dict | None = None) → Execution[source]#

delete() → None[source]#

list_executions() → list[Execution][source]#

class cloudly.gcp.workflows.WorkflowConfig[source]#

Bases: object

__init__(steps: Sequence[Step])[source]#

If your workflow requires command-line arguments, you should access individual arguments using dot notation, for example, “args.name”, “args.age”. Correspondingly in Workflow.execute(), you need to pass a dict to args, e.g. {‘name’: ‘Tom’, ‘age’: 38}.

The ‘params’ is provided whether your job needs it. If not needed, don’t provide args in Workflow.execute() and use use it in the workflow “steps”.

property definition: dict#

property workflow: Workflow#

class cloudly.gcp.workflows.Execution[source]#

Bases: object

__init__(name_or_obj: str | Execution)[source]#

property start_time: datetime#

property end_time: datetime#

property argument#

result()[source]#

status()[source]#

state() → Literal['STATE_UNSPECIFIED', 'ACTIVE', 'SUCCEEDED', 'FAILED', 'CANCELLED', 'UNAVAILABLE', 'QUEUED'][source]#

cancel()[source]#

class cloudly.gcp.workflows.Step[source]#

Bases: object

__init__(name: str, content: dict)[source]#

content is a dict of the step’s action, e.g.,:

{
  "call": "http.get",
  "args": {
    "url": "https://host.com/api1",
  },
  "result": "api_response1",
}

With “nested steps”, content will be like this:

{
  "steps": [
    {
      "step_1": {
        "call": "http.get",
        "args": {
          "url": "https://host.com/api1",
        },
        "result": "api_response1",
      },
    },
    {
      "step_2": {
        "assign": [
          {"varA": "Monday"},
          {"varB": "Tuesday"},
        ]
      }
    },
  ]
}

where each element of the list can be the output of Step.definition of some step.

property definition: dict#

class cloudly.gcp.workflows.BatchStep[source]#

Bases: Step

Running a Batch job using Workflows.

See

https://atamel.dev/posts/2023/05-30_workflows_batch_connector/ https://cloud.google.com/workflows/docs/reference/googleapis/batch/Overview https://cloud.google.com/workflows/docs/sleeping

__init__(name: str, config: JobConfig, *, delete_batch_job: bool = True)[source]#

content is a dict of the step’s action, e.g.,:

{
  "call": "http.get",
  "args": {
    "url": "https://host.com/api1",
  },
  "result": "api_response1",
}

With “nested steps”, content will be like this:

{
  "steps": [
    {
      "step_1": {
        "call": "http.get",
        "args": {
          "url": "https://host.com/api1",
        },
        "result": "api_response1",
      },
    },
    {
      "step_2": {
        "assign": [
          {"varA": "Monday"},
          {"varB": "Tuesday"},
        ]
      }
    },
  ]
}

where each element of the list can be the output of Step.definition of some step.