buildkite_sdk.group_step

  1from typing import Optional, List, Union, TypedDict
  2from buildkite_sdk.schema import PurpleStep
  3from buildkite_sdk.types import (
  4    DependsOn,
  5    NotifyEnum,
  6    StepNotify,
  7)
  8from buildkite_sdk.schema import (
  9    GroupStepClass as _group_step_class,
 10    BlockStep as _block_step,
 11    CommandStep as _command_step,
 12    InputStep as _input_step,
 13    TriggerStep as _trigger_step,
 14    WaitStep as _wait_step,
 15)
 16from .block_step import BlockStepArgs
 17from .command_step import CommandStepArgs
 18from .input_step import InputStepArgs
 19from .trigger_step import TriggerStepArgs
 20from .wait_step import WaitStepArgs
 21
 22
 23class GroupStepArgs(TypedDict):
 24    group: str
 25    steps: List[
 26        Union[
 27            BlockStepArgs,
 28            CommandStepArgs,
 29            InputStepArgs,
 30            TriggerStepArgs,
 31            WaitStepArgs,
 32        ]
 33    ]
 34    allow_dependency_failure: Optional[bool]
 35    depends_on: Optional[Union[List[Union[DependsOn, str]], str]]
 36    id: Optional[str]
 37    identifier: Optional[str]
 38    step_if: Optional[str]
 39    key: Optional[str]
 40    label: Optional[str]
 41    name: Optional[str]
 42    notify: Optional[List[Union[StepNotify, NotifyEnum]]]
 43    skip: Optional[Union[bool, str]]
 44
 45
 46def _step_to_purple_step(
 47    step: Union[
 48        _group_step_class,
 49        _block_step,
 50        _command_step,
 51        _input_step,
 52        _trigger_step,
 53        _wait_step,
 54    ],
 55):
 56    return PurpleStep(
 57        allow_dependency_failure=getattr(step, "allow_dependency_failure", None),
 58        depends_on=getattr(step, "depends_on", None),
 59        id=getattr(step, "id", None),
 60        identifier=getattr(step, "identifier", None),
 61        step_if=getattr(step, "step_if", None),
 62        key=getattr(step, "key", None),
 63        label=getattr(step, "label", None),
 64        name=getattr(step, "name", None),
 65        notify=getattr(step, "notify", None),
 66        skip=getattr(step, "skip", None),
 67        block=getattr(step, "block", None),
 68        blocked_state=getattr(step, "blocked_state", None),
 69        branches=getattr(step, "branches", None),
 70        fields=getattr(step, "fields", None),
 71        prompt=getattr(step, "prompt", None),
 72        type=None,
 73        input=getattr(step, "input", None),
 74        agents=getattr(step, "agents", None),
 75        artifact_paths=getattr(step, "artifact_paths", None),
 76        cache=getattr(step, "cache", None),
 77        cancel_on_build_failing=getattr(step, "cancel_on_build_failing", None),
 78        command=getattr(step, "command", None),
 79        commands=getattr(step, "commands", None),
 80        concurrency=getattr(step, "concurrency", None),
 81        concurrency_group=getattr(step, "concurrency_group", None),
 82        concurrency_method=getattr(step, "concurrency_method", None),
 83        env=getattr(step, "env", None),
 84        matrix=getattr(step, "matrix", None),
 85        parallelism=getattr(step, "parallelism", None),
 86        plugins=getattr(step, "plugins", None),
 87        priority=getattr(step, "priority", None),
 88        retry=getattr(step, "retry", None),
 89        signature=getattr(step, "signature", None),
 90        soft_fail=getattr(step, "soft_fail", None),
 91        timeout_in_minutes=getattr(step, "timeout_in_minutes", None),
 92        script=getattr(step, "script", None),
 93        continue_on_failure=getattr(step, "continue_on_failure", None),
 94        wait=getattr(step, "wait", None),
 95        waiter=getattr(step, "waiter", None),
 96        step_async=getattr(step, "step_async", None),
 97        build=getattr(step, "build", None),
 98        trigger=getattr(step, "trigger", None),
 99    )
100
101
102def GroupStep(
103    group: str,
104    steps: List[
105        Union[
106            _group_step_class,
107            _block_step,
108            _command_step,
109            _input_step,
110            _trigger_step,
111            _wait_step,
112        ]
113    ],
114    allow_dependency_failure: Optional[bool] = None,
115    depends_on: Optional[Union[List[Union[DependsOn, str]], str]] = None,
116    id: Optional[str] = None,
117    identifier: Optional[str] = None,
118    step_if: Optional[str] = None,
119    key: Optional[str] = None,
120    label: Optional[str] = None,
121    name: Optional[str] = None,
122    notify: Optional[List[Union[StepNotify, NotifyEnum]]] = None,
123    skip: Optional[Union[bool, str]] = None,
124) -> _group_step_class:
125    group_steps = []
126    for step in steps:
127        group_steps.append(_step_to_purple_step(step))
128
129    return _group_step_class(
130        allow_dependency_failure=allow_dependency_failure,
131        depends_on=depends_on,
132        id=id,
133        identifier=identifier,
134        step_if=step_if,
135        key=key,
136        label=label,
137        name=name,
138        notify=notify,
139        skip=skip,
140        group=group,
141        steps=group_steps,
142        block=None,
143        blocked_state=None,
144        branches=None,
145        fields=None,
146        prompt=None,
147        type=None,
148        input=None,
149        agents=None,
150        artifact_paths=None,
151        cache=None,
152        cancel_on_build_failing=None,
153        command=None,
154        commands=None,
155        concurrency=None,
156        concurrency_group=None,
157        concurrency_method=None,
158        env=None,
159        matrix=None,
160        parallelism=None,
161        plugins=None,
162        priority=None,
163        retry=None,
164        signature=None,
165        soft_fail=None,
166        timeout_in_minutes=None,
167        script=None,
168        continue_on_failure=None,
169        wait=None,
170        waiter=None,
171        step_async=None,
172        build=None,
173        trigger=None,
174    )