Last active
May 8, 2023 21:30
-
-
Save isaac-ped/3237ba78c8334cff45bc7a0242e44806 to your computer and use it in GitHub Desktop.
Automatic instantiation of single-use component resources
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
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 | |
) | |
''' |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment