Skip to content

OCI Provider Helpers

cloudspells.providers.oci.helper

OCI-specific utility helpers for the OCI provider.

Separates the OCI-API-dependent helpers from the cloud-neutral Helper class, keeping core/ free of pulumi_oci imports.

Exports

OciHelper: OCI-specific stateless utilities (image resolution, availability-domain mapping).

OciHelper

OCI-specific utility methods used by the OCI provider blocks.

All methods are stateless and safe to call multiple times. Instantiate with OciHelper() wherever needed.

Source code in packages/cloudspells-oci/src/cloudspells/providers/oci/helper.py 
 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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class OciHelper:
    """OCI-specific utility methods used by the OCI provider blocks.

    All methods are stateless and safe to call multiple times.
    Instantiate with `OciHelper()` wherever needed.
    """

    # Maps friendly OS names to (operating_system, operating_system_version)
    # tuples as expected by the OCI images API.
    _OS_MAP: ClassVar[dict[str, tuple[str, str]]] = {
        "oracle": ("Oracle Linux", "8"),
        "ubuntu": ("Canonical Ubuntu", "22.04"),
        "windows": ("Windows", "Server 2022 Standard"),
    }

    def resolve_image_id(
        self,
        compartment_id: str,
        shape: str,
        image_id: str | None = None,
        os_name: str = "oracle",
    ) -> str:
        """Resolve the compute image OCID to use for an OCI instance.

        When `image_id` is provided it is returned immediately.  Otherwise
        the method queries the OCI API for the most recently created image
        matching `os_name` that is compatible with `shape`.

        Args:
            compartment_id: OCID of the compartment to search for images.
            shape: Compute shape name used to filter compatible images
                (e.g. `"VM.Standard.E4.Flex"`).
            image_id: Optional explicit image OCID.  When provided the OCI
                API is not queried and this value is returned as-is.
            os_name: Friendly OS identifier.  Supported values:

                - `"oracle"` — Oracle Linux 8 (default)
                - `"ubuntu"` — Canonical Ubuntu 22.04
                - `"windows"` — Windows Server 2022 Standard

        Returns:
            The image OCID string to use for instance creation.

        Raises:
            ValueError: If `os_name` is not one of the supported values.

        Example:
            ```python
            ocid = OciHelper().resolve_image_id(
                compartment_id="ocid1.compartment.oc1..xxx",
                shape="VM.Standard.E4.Flex",
                os_name="oracle",
            )
            ```
        """
        if image_id is not None:
            return image_id
        if os_name not in self._OS_MAP:
            supported = ", ".join(f'"{k}"' for k in self._OS_MAP)
            raise ValueError(f"Unsupported os_name {os_name!r}. Supported values: {supported}")
        operating_system, operating_system_version = self._OS_MAP[os_name]
        import pulumi_oci as oci

        images = oci.core.get_images(
            compartment_id=compartment_id,
            operating_system=operating_system,
            operating_system_version=operating_system_version,
            shape=shape,
            sort_by="TIMECREATED",
            sort_order="DESC",
        )
        return images.images[0].id

    def get_ads(
        self,
        ads: list[dict[str, Any]],
        subnet_id: str,
    ) -> list[dict[str, str]]:
        """Convert OCI availability-domain data into OKE placement configs.

        Transforms the raw list returned by
        `oci.identity.get_availability_domains()` into the format expected
        by the OKE node pool `placement_configs` argument.

        Args:
            ads: List of availability domain dictionaries, each containing
                at least a `"name"` key (e.g. `[{"name": "AD-1"}, ...]`).
            subnet_id: Subnet OCID to assign to every placement entry.

        Returns:
            List of `{"availability_domain": str, "subnet_id": str}` dicts,
            one entry per availability domain.

        Example:
            ```python
            h = OciHelper()
            ads = [{"name": "Uocm:PHX-AD-1"}, {"name": "Uocm:PHX-AD-2"}]
            h.get_ads(ads, "ocid1.subnet.oc1...")
            # [{'availability_domain': 'Uocm:PHX-AD-1',
            #   'subnet_id': 'ocid1.subnet.oc1...'},
            #  {'availability_domain': 'Uocm:PHX-AD-2',
            #   'subnet_id': 'ocid1.subnet.oc1...'}]
            ```
        """
        return [{"availability_domain": str(ad["name"]), "subnet_id": subnet_id} for ad in ads]

resolve_image_id(compartment_id: str, shape: str, image_id: str | None = None, os_name: str = 'oracle') -> str

Resolve the compute image OCID to use for an OCI instance.

When image_id is provided it is returned immediately. Otherwise the method queries the OCI API for the most recently created image matching os_name that is compatible with shape.

Parameters:

Name Type Description Default
compartment_id str

OCID of the compartment to search for images.

 required
shape str

Compute shape name used to filter compatible images (e.g. "VM.Standard.E4.Flex").

 required
image_id str | None

Optional explicit image OCID. When provided the OCI API is not queried and this value is returned as-is.

 None
os_name str

Friendly OS identifier. Supported values:

  • "oracle" — Oracle Linux 8 (default)
  • "ubuntu" — Canonical Ubuntu 22.04
  • "windows" — Windows Server 2022 Standard
 'oracle'

Returns:

Type Description
str

The image OCID string to use for instance creation.

Raises:

Type Description
ValueError

If os_name is not one of the supported values.
Example 
ocid = OciHelper().resolve_image_id(
    compartment_id="ocid1.compartment.oc1..xxx",
    shape="VM.Standard.E4.Flex",
    os_name="oracle",
)
Source code in packages/cloudspells-oci/src/cloudspells/providers/oci/helper.py 
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
def resolve_image_id(
    self,
    compartment_id: str,
    shape: str,
    image_id: str | None = None,
    os_name: str = "oracle",
) -> str:
    """Resolve the compute image OCID to use for an OCI instance.

    When `image_id` is provided it is returned immediately.  Otherwise
    the method queries the OCI API for the most recently created image
    matching `os_name` that is compatible with `shape`.

    Args:
        compartment_id: OCID of the compartment to search for images.
        shape: Compute shape name used to filter compatible images
            (e.g. `"VM.Standard.E4.Flex"`).
        image_id: Optional explicit image OCID.  When provided the OCI
            API is not queried and this value is returned as-is.
        os_name: Friendly OS identifier.  Supported values:

            - `"oracle"` — Oracle Linux 8 (default)
            - `"ubuntu"` — Canonical Ubuntu 22.04
            - `"windows"` — Windows Server 2022 Standard

    Returns:
        The image OCID string to use for instance creation.

    Raises:
        ValueError: If `os_name` is not one of the supported values.

    Example:
        ```python
        ocid = OciHelper().resolve_image_id(
            compartment_id="ocid1.compartment.oc1..xxx",
            shape="VM.Standard.E4.Flex",
            os_name="oracle",
        )
        ```
    """
    if image_id is not None:
        return image_id
    if os_name not in self._OS_MAP:
        supported = ", ".join(f'"{k}"' for k in self._OS_MAP)
        raise ValueError(f"Unsupported os_name {os_name!r}. Supported values: {supported}")
    operating_system, operating_system_version = self._OS_MAP[os_name]
    import pulumi_oci as oci

    images = oci.core.get_images(
        compartment_id=compartment_id,
        operating_system=operating_system,
        operating_system_version=operating_system_version,
        shape=shape,
        sort_by="TIMECREATED",
        sort_order="DESC",
    )
    return images.images[0].id

get_ads(ads: list[dict[str, Any]], subnet_id: str) -> list[dict[str, str]]

Convert OCI availability-domain data into OKE placement configs.

Transforms the raw list returned by oci.identity.get_availability_domains() into the format expected by the OKE node pool placement_configs argument.

Parameters:

Name Type Description Default
ads list[dict[str, Any]]

List of availability domain dictionaries, each containing at least a "name" key (e.g. [{"name": "AD-1"}, ...]).

 required
subnet_id str

Subnet OCID to assign to every placement entry.

 required

Returns:

Type Description
list[dict[str, str]]

List of {"availability_domain": str, "subnet_id": str} dicts,
list[dict[str, str]]

one entry per availability domain.
Example 
h = OciHelper()
ads = [{"name": "Uocm:PHX-AD-1"}, {"name": "Uocm:PHX-AD-2"}]
h.get_ads(ads, "ocid1.subnet.oc1...")
# [{'availability_domain': 'Uocm:PHX-AD-1',
#   'subnet_id': 'ocid1.subnet.oc1...'},
#  {'availability_domain': 'Uocm:PHX-AD-2',
#   'subnet_id': 'ocid1.subnet.oc1...'}]
Source code in packages/cloudspells-oci/src/cloudspells/providers/oci/helper.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def get_ads(
    self,
    ads: list[dict[str, Any]],
    subnet_id: str,
) -> list[dict[str, str]]:
    """Convert OCI availability-domain data into OKE placement configs.

    Transforms the raw list returned by
    `oci.identity.get_availability_domains()` into the format expected
    by the OKE node pool `placement_configs` argument.

    Args:
        ads: List of availability domain dictionaries, each containing
            at least a `"name"` key (e.g. `[{"name": "AD-1"}, ...]`).
        subnet_id: Subnet OCID to assign to every placement entry.

    Returns:
        List of `{"availability_domain": str, "subnet_id": str}` dicts,
        one entry per availability domain.

    Example:
        ```python
        h = OciHelper()
        ads = [{"name": "Uocm:PHX-AD-1"}, {"name": "Uocm:PHX-AD-2"}]
        h.get_ads(ads, "ocid1.subnet.oc1...")
        # [{'availability_domain': 'Uocm:PHX-AD-1',
        #   'subnet_id': 'ocid1.subnet.oc1...'},
        #  {'availability_domain': 'Uocm:PHX-AD-2',
        #   'subnet_id': 'ocid1.subnet.oc1...'}]
        ```
    """
    return [{"availability_domain": str(ad["name"]), "subnet_id": subnet_id} for ad in ads]