Skip to content

Core

pulumi_extra.resource_

Utility functions for working with Pulumi resources.

References: - https://github.com/tlinhart/pulumi-aws-tags

get_resource_cls(resource_type) cached

Get the Pulumi resource class for a given resource type.

Parameters:

Name Type Description Default
resource_type str

Resource type to get the class for.

 required

Returns:

Type Description
Any | None

Pulumi resource class if found, otherwise None.
Source code in pulumi_extra/resource_.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@cache
def get_resource_cls(resource_type: str) -> Any | None:
    """Get the Pulumi resource class for a given resource type.

    Args:
        resource_type: Resource type to get the class for.

    Returns:
        Pulumi resource class if found, otherwise `None`.

    """
    try:
        _, resource = next(filter(lambda k: k[0] == resource_type, _get_resources()))
    except StopIteration:
        pulumi.log.debug(f"Resource type {resource_type} not found")
        return None

    module_name, class_name = resource
    module = import_module(module_name)
    return getattr(module, class_name)

resource_has_attribute(resource_type, attribute) cached

Check if a Pulumi resource type has a given attribute.

Parameters:

Name Type Description Default
resource_type str

Resource type to check.

 required
attribute str

Attribute to check for.

 required

Returns:

Type Description
bool

True if the resource type has the attribute, otherwise False.
Source code in pulumi_extra/resource_.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@cache
def resource_has_attribute(resource_type: str, attribute: str) -> bool:
    """Check if a Pulumi resource type has a given attribute.

    Args:
        resource_type: Resource type to check.
        attribute: Attribute to check for.

    Returns:
        `True` if the resource type has the attribute, otherwise `False`.
    """
    cls = get_resource_cls(resource_type)
    if cls is None:
        msg = f"Unable to resolve resource type {resource_type!r}"
        raise UnknownResourceTypeError(msg)

    sig = signature(cls._internal_init)
    return attribute in sig.parameters

pulumi_extra.stack_reference

Utils for stack references.

get_stack_outputs(*refs) 

get_stack_outputs(ref: str) -> pulumi.Output[Any]

get_stack_outputs(*refs: str) -> list[pulumi.Output[Any]]

Get outputs from a output reference shorthands. Supports brace expansion.

  • Single output reference: ("<stack_ref>:<output_key>").
  • Multiple outputs using brace expansion: ("<stack_ref>:{<output_key_1>,<output_key_2>}").

Parameters:

Name Type Description Default
*refs str

Output references.

 ()
Source code in pulumi_extra/stack_reference.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def get_stack_outputs(  # type: ignore[misc]
    *refs: str,
) -> pulumi.Output[Any] | list[pulumi.Output[Any]]:
    """Get outputs from a output reference shorthands. Supports brace expansion.

    - Single output reference: (`"<stack_ref>:<output_key>"`).
    - Multiple outputs using brace expansion: (`"<stack_ref>:{<output_key_1>,<output_key_2>}"`).

    Args:
        *refs: Output references.

    """
    outputs = _get_stack_outputs(*refs)
    output_values = list(outputs.values())
    if len(output_values) == 1:
        return output_values[0]

    return output_values

get_stack_reference(ref) cached

Resolve given stack reference shorthand to fully qualified stack reference.

The shorthand can be one of the following:

  • "{stack}"

    Returns the stack reference for the current project and organization.

  • "{project}/{stack}"

    Returns the stack reference for the current organization.

  • "{organization}/{project}/{stack}"

    No change is made to the stack reference.

Source code in pulumi_extra/stack_reference.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@cache
def get_stack_reference(ref: str) -> pulumi.StackReference:
    """Resolve given stack reference shorthand to fully qualified stack reference.

    The shorthand can be one of the following:

    - `"{stack}"`

        Returns the stack reference for the current project and organization.

    - `"{project}/{stack}"`

        Returns the stack reference for the current organization.

    - `"{organization}/{project}/{stack}"`

        No change is made to the stack reference.

    """
    fqr = _resolve_stack_ref(ref)
    return pulumi.StackReference(fqr)

re_export(*refs)

Re-export outputs from a output reference shorthands.

Parameters:

Name Type Description Default
*refs str

Output references.

 ()
Source code in pulumi_extra/stack_reference.py 
83
84
85
86
87
88
89
90
91
92
def re_export(*refs: str) -> None:
    """Re-export outputs from a output reference shorthands.

    Args:
        *refs: Output references.

    """
    outputs = _get_stack_outputs(*refs)
    for (_, output_key), output in outputs.items():
        pulumi.export(output_key, output)

pulumi_extra.output

Utils for outputs.

render_template(template, *, context=None, inputs=None) 

render_template(
    template: Path | str, *, context: Mapping[str, Any]
) -> str

render_template(
    template: Path | str,
    *,
    inputs: Mapping[str, pulumi.Input[Any]],
) -> pulumi.Output[str]

Render a template file with the given context.

Parameters:

Name Type Description Default
template Path | str

The template file or inline template string.

 required
context Mapping[str, Any] | None

The context to render the template with. Conflicts with inputs.

 None
inputs Mapping[str, Input[Any]] | None

The inputs to render the template with. Conflicts with context.

 None
Source code in pulumi_extra/output.py 
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
def render_template(
    template: Path | str,
    *,
    context: Mapping[str, Any] | None = None,
    inputs: Mapping[str, pulumi.Input[Any]] | None = None,
) -> str | pulumi.Output[str]:
    """Render a template file with the given context.

    Args:
        template: The template file or inline template string.
        context: The context to render the template with. Conflicts with inputs.
        inputs: The inputs to render the template with. Conflicts with context.
    """
    if isinstance(template, Path):
        template = template.read_text()

    jinja_tpl = Template(template, undefined=StrictUndefined)

    # Render with Python values.
    if context is not None and inputs is None:
        return jinja_tpl.render(context)

    # Render with Pulumi inputs.
    if context is None and inputs is not None:
        return pulumi.Output.all(inputs).apply(lambda args: jinja_tpl.render(args[0]))

    # Only one of context or inputs must be provided.
    msg = "Either context or input must be provided."
    raise ValueError(msg)

pulumi_extra.transforms

override_default_provider(*rt_or_it, provider)

Override the default provider for resources and invokes of given types.

Parameters:

Name Type Description Default
*rt_or_it str

Resource types or invoke tokens to match.

 ()
provider ProviderResource

Provider to override.

 required
Source code in pulumi_extra/transforms/runtime.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
def override_default_provider(
    *rt_or_it: str,
    provider: pulumi.ProviderResource,
) -> None:
    """Override the default provider for resources and invokes of given types.

    Args:
        *rt_or_it: Resource types or invoke tokens to match.
        provider: Provider to override.

    """
    pulumi.runtime.register_resource_transform(override_resource_options(*rt_or_it, provider=provider))
    pulumi.runtime.register_invoke_transform(override_invoke_options(*rt_or_it, provider=provider))

override_invoke(*invoke_tokens, args=None, opts=None)

Pulumi transform factory for invoke tokens (get_*).

Parameters:

Name Type Description Default
*invoke_tokens str

Invoke tokens to match. Supports glob patterns and brace expand.

 ()
args _Args | Callable[[_Args], _Args] | None

Invoke arguments to override, or a callable that returns the new arguments from given args.args input.

 None
opts InvokeOptions | Callable[[InvokeOptions], InvokeOptions] | None

Invoke options to override, or a callable that returns the new options given args.opts input.

 None
Source code in pulumi_extra/transforms/invoke.py 
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
def override_invoke(
    *invoke_tokens: str,
    args: _Args | Callable[[_Args], _Args] | None = None,
    opts: pulumi.InvokeOptions | Callable[[pulumi.InvokeOptions], pulumi.InvokeOptions] | None = None,
) -> pulumi.InvokeTransform:
    """Pulumi transform factory for invoke tokens (`get_*`).

    Args:
        *invoke_tokens: Invoke tokens to match. Supports glob patterns and brace expand.
        args: Invoke arguments to override, or a callable that returns the new arguments from given `args.args` input.
        opts: Invoke options to override, or a callable that returns the new options given `args.opts` input.

    """
    args_ = args

    def transform(args: pulumi.InvokeTransformArgs) -> pulumi.InvokeTransformResult | None:
        nonlocal args_, opts

        for it in chain.from_iterable(map(braceexpand, invoke_tokens)):
            if not fnmatch(args.token, it):
                continue

            # Transform invoke arguments
            if TYPE_CHECKING:
                assert isinstance(args.args, dict)

            if callable(args_):  # noqa: SIM108
                new_args = args_(args.args)
            else:
                new_args = args.args | args_ if args_ is not None else args.args

            # Transform invoke options
            new_opts = opts(args.opts) if callable(opts) else (opts or pulumi.InvokeOptions())
            new_opts = pulumi.InvokeOptions.merge(args.opts, new_opts)

            return pulumi.InvokeTransformResult(args=new_args, opts=new_opts)

        return None

    return transform

override_invoke_defaults(*invoke_tokens, defaults)

Pulumi transform factory that provides default arguments to matching invoke tokens.

Parameters:

Name Type Description Default
*invoke_tokens str

Invoke tokens to match.

 ()
defaults dict[str, Any]

Default arguments.

 required
Source code in pulumi_extra/transforms/invoke.py 
61
62
63
64
65
66
67
68
69
70
71
72
def override_invoke_defaults(*invoke_tokens: str, defaults: dict[str, Any]) -> pulumi.InvokeTransform:
    """Pulumi transform factory that provides default arguments to matching invoke tokens.

    Args:
        *invoke_tokens: Invoke tokens to match.
        defaults: Default arguments.

    """
    return override_invoke(
        *invoke_tokens,
        args=lambda args: defaults | args,
    )

override_invoke_options(*invoke_tokens, **options)

Pulumi transform factory that overrides the invoke options for matching invoke tokens.

Parameters:

Name Type Description Default
*invoke_tokens str

Invoke tokens to match.

 ()
options Any

Arguments of pulumi.InvokeOptions.

 {}
Source code in pulumi_extra/transforms/invoke.py 
75
76
77
78
79
80
81
82
83
84
85
86
def override_invoke_options(*invoke_tokens: str, **options: Any) -> pulumi.InvokeTransform:
    """Pulumi transform factory that overrides the invoke options for matching invoke tokens.

    Args:
        *invoke_tokens: Invoke tokens to match.
        options: Arguments of `pulumi.InvokeOptions`.

    """
    return override_invoke(
        *invoke_tokens,
        opts=pulumi.InvokeOptions(**options),
    )

override_resource(*resource_types, props=None, opts=None)

Pulumi transform factory for resources.

Parameters:

Name Type Description Default
*resource_types str

Resource types to match. Supports glob patterns and brace expand.

 ()
props _Props | Callable[[_Props], _Props] | None

Resource properties to override, or a callable that returns the new properties from given args.props input.

 None
opts ResourceOptions | Callable[[ResourceOptions], ResourceOptions] | None

Resource options to override, or a callable that returns the new options given args.opts input.

 None
Source code in pulumi_extra/transforms/resource_.py 
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
def override_resource(
    *resource_types: str,
    props: _Props | Callable[[_Props], _Props] | None = None,
    opts: pulumi.ResourceOptions | Callable[[pulumi.ResourceOptions], pulumi.ResourceOptions] | None = None,
) -> pulumi.ResourceTransform:
    """Pulumi transform factory for resources.

    Args:
        *resource_types: Resource types to match. Supports glob patterns and brace expand.
        props: Resource properties to override, or a callable that returns the new properties from given `args.props` input.
        opts: Resource options to override, or a callable that returns the new options given `args.opts` input.

    """  # noqa: E501

    def transform(args: pulumi.ResourceTransformArgs) -> pulumi.ResourceTransformResult | None:
        nonlocal props, opts

        for rt in chain.from_iterable(map(braceexpand, resource_types)):
            if not fnmatch(args.type_, rt):
                continue

            # Transform resource properties
            if TYPE_CHECKING:
                assert isinstance(args.props, dict)

            if callable(props):
                new_props = props(args.props)
            else:
                new_props = args.props | props if props is not None else args.props

            # Transform resource options
            new_opts = opts(args.opts) if callable(opts) else (opts or pulumi.ResourceOptions())
            new_opts = pulumi.ResourceOptions.merge(args.opts, new_opts)

            return pulumi.ResourceTransformResult(props=new_props, opts=new_opts)

        return None

    return transform

override_resource_defaults(*resource_types, defaults)

Pulumi transform factory that provides default properties to matching resource types.

Parameters:

Name Type Description Default
*resource_types str

Resource type to match.

 ()
defaults dict[str, Input[Any]]

Default properties.

 required
Source code in pulumi_extra/transforms/resource_.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def override_resource_defaults(
    *resource_types: str,
    defaults: dict[str, pulumi.Input[Any]],
) -> pulumi.ResourceTransform:
    """Pulumi transform factory that provides default properties to matching resource types.

    Args:
        *resource_types: Resource type to match.
        defaults: Default properties.

    """
    return override_resource(
        *resource_types,
        props=lambda props: defaults | props,
    )

override_resource_options(*resource_types, **options)

Pulumi transform factory that overrides the resource options for resources of given types.

Parameters:

Name Type Description Default
*resource_types str

Resource types to match.

 ()
options Any

Arguments of pulumi.ResourceOptions.

 {}
Source code in pulumi_extra/transforms/resource_.py 
73
74
75
76
77
78
79
80
81
82
83
84
def override_resource_options(*resource_types: str, **options: Any) -> pulumi.ResourceTransform:
    """Pulumi transform factory that overrides the resource options for resources of given types.

    Args:
        *resource_types: Resource types to match.
        options: Arguments of `pulumi.ResourceOptions`.

    """
    return override_resource(
        *resource_types,
        opts=pulumi.ResourceOptions(**options),
    )