Automatic instantiation of single-use component resources

ResourceGroup.py

"""
Automatically define singleton resource groups

The structure of the python pulumi library is such that many properties
of resources are not easily modifyable after initialization.

This module functions by delaying the creation of resource attributes
within the ResourceGroup class until the class has been fully instantiated.
"""
from typing import Any, ClassVar, Iterable, TypeVar, Generic
from types import ModuleType
import pulumi

R = TypeVar("R", bound=pulumi.Resource, covariant=True)

created_resources: dict["DelayedResource[pulumi.Resource]", pulumi.Resource] = {}


class DelayedResourceAttribute:
    """Delay access to attributes on DelayedResources
    
    Once the component resource has been determined, these attributes can be resolved with
    attr.resolve(component)
    which will also recursively resolve any other necessary resource.

    """

    def __init__(self, parent: "Resolvable", attr: str, call_args: Iterable[Any] = [], call_kwargs: dict[str, Any] = {}):
        self.parent = parent
        self.attr = attr
        self.call_args = call_args
        self.call_kwargs = call_kwargs

    def __getattr__(self, key: str):
        return DelayedResourceAttribute(self, key)

    def __str__(self):
        output = f"{self.parent}"
        output += f".{self.attr}"
        call_str = ''
        if self.call_args:
            call_str += ','.join([str(x) for x in self.call_args])
        if self.call_kwargs:
            call_str += ','.join([f"{k}={v}" for k, v in self.call_kwargs])
        if call_str:
            output += f"({call_str})"
        return output

    def __call__(self, *args: Any, **kwargs: Any):
        return DelayedResourceAttribute(self,"__call__", args, kwargs)

    def resolve(self, resource: pulumi.ComponentResource) -> Any:
        resolved = self.parent.resolve(resource)
        resolved = getattr(resolved, self.attr)
        if self.call_args or self.call_kwargs:
            resolved = resolved(*self.call_args, **self.call_kwargs)
        return resolved

class DelayedResource(Generic[R]):
    """Delay the instantiation of a resource until `create()` is called

    This allows us to inject the parent into the resource's options
    """

    def __init__(self, wrapped: type[R]):
        self.wrapped = wrapped
        self._created = None

    def __call__(self, *args: Any, **kwargs: Any):
        self.args = args
        self.kwargs = kwargs
        return self

    def __getattr__(self, key: str) -> Any:
        return DelayedResourceAttribute(self, key)

    @staticmethod
    def _resolve_arg(arg: Any, resource: pulumi.ComponentResource) -> Any:
        """Recursively resolves an argument to the resource constructor
        
        This way if it references other resources it can still be used.
        """
        if isinstance(arg, DelayedResourceAttribute):
            return arg.resolve(resource)
        if isinstance(arg, list):
            args : list[Any] = arg
            return [DelayedResource._resolve_arg(item, resource) for item in args]
        if isinstance(arg, dict):
            kwargs : dict[str, Any] = arg
            return {key: DelayedResource._resolve_arg(value, resource) for key, value in kwargs.items()}
        return arg
        
    def resolve(self, parent: pulumi.ComponentResource) -> R:
        """Instantiate the resource with the parent option injected"""
        if self._created:
            return self._created

        args = [self._resolve_arg(arg, parent) for arg in self.args]
        kwargs = {key: self._resolve_arg(value, parent) for key, value in self.kwargs.items()}

        opts = self.kwargs.get("opts", pulumi.ResourceOptions())
        kwargs["opts"] = pulumi.ResourceOptions.merge(
            opts, pulumi.ResourceOptions(parent=parent)
        )

        self._created = self.wrapped(*args, **kwargs)

        return self._created

Resolvable = DelayedResourceAttribute | DelayedResource[pulumi.Resource]

class ModuleDelayer:
    """Wraps a module so that resources referenced within it are delayed"""

    def __init__(self, wrapped: ModuleType):
        self.__wrapped = wrapped
        self._child = None

    def __getattr__(self, __name: str) -> Any:
        attr = getattr(self.__wrapped, __name)
        if isinstance(attr, type) and issubclass(attr, pulumi.Resource):
            return DelayedResource(attr)
        if isinstance(attr, ModuleType):
            return ModuleDelayer(attr)
        return attr

class AttrDelayer(dict[str, Any]):
    """Wraps a namespace so that resources referenced within it are delayed.

    This is used as the base namespace for a ResourceGroup class,
    so that the creation of class variables defined within it are appropriately delayed.
    """

    def __init__(self, mapping: dict[str, DelayedResource[pulumi.Resource]]):
        self.__mapping = mapping
        super().__init__()

    def __getitem__(self, key: str) -> Any:
        if key in self.__mapping:
            return self.__mapping[key]
        try:
            return super().__getitem__(key)
        except KeyError:
            attr = globals().get(key)
            if isinstance(attr, ModuleType):
                return ModuleDelayer(attr)
            raise

    def __setitem__(self, key: str, value: Any):
        if isinstance(value, DelayedResource):
            self.__mapping[key] = value
        super().__setitem__(key, value)


class DelayedResourceMeta(type):
    """The metaclass for ResourceGroup - specifies the AttrDelayer namespace"""
    # Maps class names onto the resources that they define
    resources: ClassVar[
        dict[str, dict[str, DelayedResource[pulumi.Resource]]]
    ] = {}

    @classmethod
    def __prepare__(cls, classname: str, _bases: Any, **_):
        cls.resources[classname] = {}
        return AttrDelayer(cls.resources[classname])

RG = TypeVar("RG", bound="ResourceGroup")
class ResourceGroup(pulumi.ComponentResource, metaclass=DelayedResourceMeta):
    """Subclasses of ResourceGroup will automatically create a ComponentResource,
    and automatically mark all pulumi.Resource classvars as children of that class.

    Arguments to the subclass should be passed as arguments to the class constructor, e.g.
    class MyResourceGroup(ResourceGroup, type_="myorg:CustomGroup", name="myGroup"):
    """

    _instance: "ResourceGroup | None" = None

    @classmethod
    def get(cls: type[RG]) -> RG:
        assert cls._instance is not None
        return cls._instance  # type: ignore

    @property
    def _resources(self) -> dict[str, Any]:
        name = type(self).__name__
        return DelayedResourceMeta.resources[name]

    def __init_subclass__(
        cls: type[RG],
        /,
        type_: str | None = None,
        name: str | None = None,
        opts: pulumi.ResourceOptions | None = None,
    ) -> None:
        super().__init_subclass__()
        if name is None:
            name = cls.__name__
        if type_ is None:
            type_ = f"ResourceGroup:{name}"

        setattr(cls, "_instance", cls(type_, name, opts))


    def __init__(self, type_: str, name: str, opts: pulumi.ResourceOptions | None):
        if self._instance is not None:
            raise RuntimeError(
                "Cannot create more than one instance of a ResourceGroup"
            )
        super().__init__(type_, name, opts=opts)
        resources = {}
        for name, resource in self._resources.items():
            resources[name] = resource.resolve(self)
        self.register_outputs(resources)

'''
# ####
# # Example usage
# ####
#
import pulumi_gcp as gcp

project_id = "my-project"
region = "us-west1"

class NetworkGroup(
    ResourceGroup,
):



    # The network for the k8 cluster
    network = gcp.compute.Network(
        resource_name="vpc",
        project=project_id,
        auto_create_subnetworks=False,
        routing_mode="GLOBAL"
    )


    # We're setting up a cluster with VPC Native networking and alias ips.
    # Info on picking ip ranges:
    # https://cloud.google.com/kubernetes-engine/docs/concepts/alias-ips#defaults_limits
    subnetwork = gcp.compute.Subnetwork(
        resource_name="subnet",
        project=project_id,
        region=region,
        network="badeep",
        ip_cidr_range="10.0.0.0/20",
        secondary_ip_ranges=[
            gcp.compute.SubnetworkSecondaryIpRangeArgs(
                range_name="pod-range",
                # /14 ~ max 1024 nodes, 112640 pods
                # 10.20.0.1 - 10.23.255.254
                ip_cidr_range="10.20.0.0/14",
            ),
            gcp.compute.SubnetworkSecondaryIpRangeArgs(
                range_name="service-range",
                # /24 ~ max 256 services
                # 10.24.0.1 - 10.24.0.254
                ip_cidr_range="10.24.0.0/24",
            ),
        ],
        opts=pulumi.ResourceOptions(
            delete_before_replace=True,
        ),
    )

group = NetworkGroup.get()

policy = gcp.organizations.get_iam_policy(
    bindings=[
        gcp.organizations.GetIAMPolicyBindingArgs(
            role="roles/artifactregistry.writer",
            members=[],
        ),
        gcp.organizations.GetIAMPolicyBindingArgs(
            role="roles/artifactregistry.reader",
            members=[],
        ),
    ],
)
class ArtifactRegistry(
    ResourceGroup
):
    repository = gcp.artifactregistry.Repository(
        resource_name="boop",
        repository_id="beep",
        location="US",
        format="DOCKER",
    )

    iam_policy = gcp.artifactregistry.RepositoryIamPolicy(
        resource_name=f"noop-iam-policy",
        location="US",
        repository="beep",
        policy_data=policy.policy_data
    )
'''