Source code for gcip2.pipeline.pipeline

 1import dataclasses
 2from collections.abc import Iterable
 3from typing import Self
 4
 5import injector
 6
 7from gcip2.pipeline_core import (
 8    Default,
 9    Image,
10    Job,
11    JobBuilderImpl,
12    Pipeline,
13    Stage,
14    Workflow,
15    WorkflowAutoCancel,
16    WorkflowAutoCancelOnNewCommit,
17    WorkflowRule,
18    WorkflowWhen,
19)
20from gcip2.project_config import ProjectConfig
21
22

[docs]
23class _PipelineBuilder:

[docs]
24    def build(self) -> Pipeline:
25        raise NotImplementedError


26
27

[docs]
28@injector.inject
29@dataclasses.dataclass
30class _BasePipelineBuilder(_PipelineBuilder):
31    _config: ProjectConfig
32    _di: injector.Injector
33
34    model: Pipeline = dataclasses.field(init=False, default_factory=Pipeline)
35
36    def __post_init__(self):
37        "extension point for childs"
38

[docs]
39    def apply(self: Self) -> Self:
40        return self

41

[docs]
42    def build(self: Self) -> Pipeline:
43        if not self.model.default:
44            self.model.default = Default()
45
46        if not self.model.default.image:
47            image_data = self._config.compose.images.get("base_python")
48            if not image_data:
49                raise ValueError(f"image_data not found in {self._config.compose.path.name}")
50            self.model.default.image = Image(name=image_data.image)
51
52        if not self.model.default.tags:
53            self.model.default.tags = ["static-k8s"]
54
55        if not self.model.stages:
56            self.model.stages = [Stage.JOBS]
57
58        return self.model.model_copy(deep=True)

59

[docs]
60    def job(self, job_class: type[JobBuilderImpl]) -> JobBuilderImpl:
61        return self._di.create_object(job_class)

62

[docs]
63    def add_jobs(self, jobs: Iterable[Job | JobBuilderImpl]):
64        self.model.jobs.extend(jobs)

65

[docs]
66    def with_workflow(self, workflow: Workflow):
67        self.model.workflow = workflow
68        return self

69

[docs]
70    def with_default(self, default: Default):
71        self.model.default = default
72        return self


73
74

[docs]
75class PipelineBuilderImpl(_BasePipelineBuilder):
76    def __post_init__(self) -> None:
77        self.model.workflow = Workflow(
78            auto_cancel=WorkflowAutoCancel(on_new_commit=WorkflowAutoCancelOnNewCommit.NONE),
79            rules=[
80                WorkflowRule(
81                    if_='$CI_PIPELINE_SOURCE == "push" && $CI_OPEN_MERGE_REQUESTS',
82                    when=WorkflowWhen.NEVER,
83                ),
84                WorkflowRule(
85                    when=WorkflowWhen.ALWAYS,
86                ),
87            ],
88        )

89
90

[docs]
91class GitlabCiBuilderImpl(_BasePipelineBuilder):
92    pass