1import importlib.util
2import os
3import pathlib
4import sys
5from difflib import unified_diff
6from enum import Enum
7from typing import Any, Optional, Self, TypeVar
8
9import injector
10import yaml
11
12from gcip2.ci.jobs import trigger
13from gcip2.logging import logger as LOGGER
14from gcip2.pipeline.dumper import CustomDumper
15from gcip2.pipeline.pipeline import GitlabCiBuilderImpl, PipelineBuilderImpl
16from gcip2.pipeline_core import JobBuilderImpl, Pipeline, Stage, WorkflowRule, WorkflowWhen
17
18T = TypeVar("T", PipelineBuilderImpl, GitlabCiBuilderImpl)
19
20
[docs]
21class TriggerPipelineDefaults(str, Enum):
22 out_file = "out/pipeline.gitlab-ci.yml"
23
24
[docs]
25class PipelineBuilder:
26 yaml_dumper = CustomDumper
27
28 @injector.inject
29 def __init__(self, di: injector.Injector) -> None:
30 self._di = di
31
[docs]
32 @staticmethod
33 def show_file_diff(path: pathlib.Path, data: str):
34 old_data = path.read_text() if path.exists() else ""
35
36 diff = unified_diff(
37 old_data.splitlines(),
38 data.splitlines(),
39 fromfile="old",
40 tofile="new",
41 lineterm="",
42 )
43
44 diff_data = [s for s in diff]
45 diff_string = "\n".join([f"pipeline: {path}"] + diff_data)
46
47 LOGGER.warning(diff_string) if diff_data else LOGGER.info(f"pipeline: {path} has no changes")
48
[docs]
49 def load_pipeline(
50 self: Self,
51 path: Any,
52 obj_type: type[T],
53 ) -> T:
54 path = pathlib.Path(path)
55 spec: Any = importlib.util.spec_from_file_location(
56 "user_pipeline",
57 path,
58 )
59
60 module = importlib.util.module_from_spec(spec)
61 path_str = str(path.parent)
62 sys.path.insert(0, path_str)
63 try:
64 spec.loader.exec_module(module)
65 finally:
66 sys.path.remove(path_str)
67
68 for obj in module.__dict__.values():
69 if isinstance(obj, type) and issubclass(obj, obj_type) and obj is not obj_type:
70 return self._di.create_object(obj)
71
72 raise RuntimeError(f"{obj_type.__name__} child class not found")
73
[docs]
74 def render_pipeline(self: Self, pipeline: Pipeline) -> dict[str, Any]:
75 pipeline = Pipeline.model_validate(pipeline, strict=True)
76 result: dict[str, Any] = pipeline.dump(exclude=["jobs"])
77
78 for job in pipeline.jobs:
79 if isinstance(job, JobBuilderImpl):
80 job = job.build()
81 if not job.name:
82 raise ValueError("Job name is missing.")
83 if job.name in result.keys():
84 raise ValueError("Job with same name already added.")
85 result[job.name] = job.dump(exclude=["name"])
86 return result
87
[docs]
88 def build_pipeline_file(
89 self: Self,
90 pipeline: Pipeline,
91 path: Optional[pathlib.Path],
92 **_: Any,
93 ) -> None:
94 os.makedirs("out", exist_ok=True)
95 pipeline_copy = pipeline.model_copy(deep=True)
96
97 data = yaml.dump(
98 self.render_pipeline(pipeline=pipeline_copy),
99 sort_keys=False,
100 allow_unicode=True,
101 Dumper=CustomDumper,
102 )
103
104 path = path or pathlib.Path(os.getcwd()) / TriggerPipelineDefaults.out_file.value
105
106 self.show_file_diff(path=path, data=data)
107
108 path.write_text(data)
109
[docs]
110 def build_gitlab_ci(
111 self: Self,
112 ci_file_path: pathlib.Path,
113 out_gitlab_ci: pathlib.Path,
114 ) -> None:
115 _build_trigger_pipeline = self._di.create_object(trigger.BuildTriggerPipeline)
116 _trigger_pipeline = self._di.create_object(trigger.TriggerPipeline)
117
118 pipeline_entry = self.load_pipeline(ci_file_path, obj_type=GitlabCiBuilderImpl)
119
120 pipeline_obj: Pipeline = pipeline_entry.apply().build()
121
122 downstream_rule = WorkflowRule(
123 if_='$CI_PIPELINE_SOURCE == "parent_pipeline"',
124 when=WorkflowWhen.ALWAYS,
125 )
126
127 if not pipeline_obj.jobs:
128 pipeline_obj.jobs = [
129 _build_trigger_pipeline.apply(),
130 _trigger_pipeline.apply(),
131 ]
132
133 if pipeline_obj.workflow and pipeline_obj.workflow.rules and downstream_rule not in pipeline_obj.workflow.rules:
134 pipeline_obj.workflow.rules = [downstream_rule] + pipeline_obj.workflow.rules
135
136 if not pipeline_obj.stages:
137 pipeline_obj.stages = [Stage.JOBS]
138
139 self.build_pipeline_file(
140 pipeline=pipeline_obj,
141 path=out_gitlab_ci,
142 )
143
[docs]
144 def build_pipeline(
145 self: Self,
146 ci_file_path: pathlib.Path,
147 out_pipeline_path: pathlib.Path,
148 ) -> None:
149 pipeline_entry: PipelineBuilderImpl = self.load_pipeline(ci_file_path, obj_type=PipelineBuilderImpl)
150
151 pipeline_obj: Pipeline = pipeline_entry.apply().build()
152
153 downstream_rule = WorkflowRule(
154 if_='$CI_PIPELINE_SOURCE == "parent_pipeline"',
155 when=WorkflowWhen.ALWAYS,
156 )
157
158 if pipeline_obj.workflow and pipeline_obj.workflow.rules and downstream_rule not in pipeline_obj.workflow.rules:
159 pipeline_obj.workflow.rules = [downstream_rule] + pipeline_obj.workflow.rules
160
161 self.build_pipeline_file(
162 pipeline=pipeline_obj,
163 path=out_pipeline_path,
164 )