cf-python-client/cloudfoundry_client/v3/buildpacks.py at bcd5b3ed5faedd90b0c9248b633791b3a0dde7ab · cloudfoundry-community/cf-python-client · 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
from typing import TYPE_CHECKING

from cloudfoundry_client.v3.entities import EntityManager, Entity

if TYPE_CHECKING:
    from cloudfoundry_client.client import CloudFoundryClient


class BuildpackManager(EntityManager[Entity]):
    def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
        super().__init__(target_endpoint, client, "/v3/buildpacks")

    def create(
        self,
        name: str,
        position: int | None = 0,
        enabled: bool | None = True,
        locked: bool | None = False,
        stack: str | None = None,
        meta_labels: dict | None = None,
        meta_annotations: dict | None = None,
    ) -> Entity:
        data = {
            "name": name,
            "position": position,
            "enabled": enabled,
            "locked": locked,
            "stack": stack,
        }
        self._metadata(data, meta_labels, meta_annotations)
        return super()._create(data)

    def remove(self, buildpack_guid: str, asynchronous: bool = True) -> str | None:
        return super()._remove(buildpack_guid, asynchronous)

    def update(
        self,
        buildpack_guid: str,
        name: str,
        position: int | None = 0,
        enabled: bool | None = True,
        locked: bool | None = False,
        stack: str | None = None,
        meta_labels: dict | None = None,
        meta_annotations: dict | None = None,
    ) -> Entity:
        data = {
            "name": name,
            "position": position,
            "enabled": enabled,
            "locked": locked,
            "stack": stack,
        }
        self._metadata(data, meta_labels, meta_annotations)
        return super()._update(buildpack_guid, data)

    def upload(self, buildpack_guid: str, buildpack_zip: str, asynchronous: bool = False) -> Entity:
        buildpack = super()._upload_bits(buildpack_guid, buildpack_zip)
        if not asynchronous:
            self.client.v3.jobs.wait_for_job_completion(buildpack.job()["guid"])
            buildpack_after_job = super().get(buildpack["guid"])
            buildpack_after_job["links"]["job"] = buildpack["links"]["job"]
            buildpack_after_job.job = buildpack.job
            return buildpack_after_job
        return buildpack