image-stream/stream.py at master · xiaobao520123/image-stream · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import argparse
from yaml import safe_load

from image_stream.global_config import GlobalConfig
from image_stream.task import *


def main():
    parser = argparse.ArgumentParser(
        description="image-stream allows you to tag and copy docker images to a specified registry."
    )
    parser.add_argument(
        "configuration",
        help="Path to the configuration file containing the registry and image details.",
    )
    args = parser.parse_args()

    task_config = load_config(args.configuration)
    tasks = build_tasks(task_config)
    deliver(tasks)


def load_config(file_path):
    with open(file_path, "r") as file:
        config = safe_load(file)

    config = config.get("image-stream")
    if config is None:
        raise ValueError(
            "Invalid configuration format. key: 'image-stream' is required."
        )

    GlobalConfig.registry = config.get("registry", None)
    GlobalConfig.repository = config.get("repository", None)
    GlobalConfig.platform = config.get("platform", None)
    GlobalConfig.print()
    return config.get("images", [])


def build_tasks(task_configs):
    tasks = []
    for task in task_configs:
        name = task.get("name")

        source_type = task.get("source")
        source = None
        if source_type == Source.image:
            source = task.get("image")
            source = Image(
                registry=source.get("registry"),
                repository=source.get("repository"),
                tag=source.get("tag"),
                digest=source.get("digest", None),
                platform=source.get("platform", GlobalConfig.platform),
            )
        elif source_type == Source.helm:
            helm = task.get("helm")
            image_config = helm.get("image")
            image_config = HelmChartImageConfig(
                registry=image_config.get("registry"),
                repository=image_config.get("repository"),
                tag=image_config.get("tag"),
                digest=image_config.get("digest", None),
                platform=image_config.get("platform", None),
            )
            source = HelmChart(
                repository=helm.get("repository"),
                chart=helm.get("chart"),
                image_config=image_config,
                version=helm.get("version"),
            )
        else:
            raise ValueError(f"Invalid source: {source_type}")

        delivery = task.get("delivery")
        if not delivery:
            raise ValueError(f"{name} key 'delivery' not found.")

        delivery = Delivery(
            registry=delivery.get("registry", GlobalConfig.registry),
            repository=delivery.get("repository", GlobalConfig.repository),
            tag=delivery.get("tag"),
            platform=delivery.get("platform", GlobalConfig.platform),
        )

        task = Task(
            name=name,
            source=source,
            delivery=delivery,
        )

        tasks.append(task)
    return tasks


def deliver(tasks):
    for task in tasks:
        print(f"Start streaming task: {task.name}")
        task.deliver()
        print("")


if __name__ == "__main__":
    main()