Typing Pulumi Component Inputs and Outputs

A ComponentResource is only as reusable as its signature. A constructor that takes **kwargs and stores results on attributes invented halfway down __init__ forces every consumer to read your source to learn the contract, and it lets a typo survive until pulumi up fails against a live account. This guide belongs to Pulumi Component Resources within Pulumi Patterns & Provider Management, and it shows how to declare a component's interface so an editor can autocomplete it and mypy can reject misuse in CI.

Context

Pulumi's Python SDK exposes two generic aliases that carry the whole contract. pulumi.Output[T] is the future-like wrapper the engine resolves once a resource exists; it also carries the dependency edge between the resource that produced the value and anything that consumes it. pulumi.Input[T] is defined as Union[T, Awaitable[T], Output[T]] — the set of things a caller is allowed to hand you for a value that will eventually reach a provider.

The distinction matters because the two most common call sites look different. One caller passes a literal from stack configuration; another passes vpc.id from a resource created moments earlier. If the argument is annotated str, the second caller is a type error even though it is the more idiomatic Pulumi code. If it is annotated Output[str], the first caller has to write pulumi.Output.from_input("prod") for no reason. Input[str] accepts both, which is exactly why every generated provider SDK uses it.

Prerequisites

• Python 3.9+ with pulumi >= 3.0 and pulumi-aws >= 6.0 pinned in the lockfile.
• mypy >= 1.8 installed in the same virtual environment as the SDKs, so it can read their bundled type information.
• A component you already know how to wire, such as the one from building a reusable VPC component.
• Familiarity with Output resolution and apply, since a typed surface is mostly a surface of Output values.

# CLI: confirm both SDKs ship inline types rather than stub-only packages
python -c "import pulumi, pulumi_aws, pathlib; print(pathlib.Path(pulumi.__file__).parent / 'py.typed')"
mypy --version

Implementation

The worked example is a queue-worker component: an SQS queue, a dead-letter queue, an IAM role, a Lambda function, an event source mapping, and a log group. It has enough surface to show every rule.

1. Accept Input[T], and a plain type only when Python must read the value

Annotate anything that is forwarded to a provider argument as Input[T]. Annotate as a plain int or str only the values your own Python code branches on, loops over, or interpolates into a resource name — an Output cannot be compared, iterated, or formatted at construction time, so a plain type is a truthful declaration that the value must be known before the graph is built.

# components/queue_worker/args.py
# CLI: python -c "from components.queue_worker.args import QueueWorkerArgs; print(QueueWorkerArgs.__doc__)"
from dataclasses import dataclass, field
from typing import Mapping, Optional

import pulumi


@dataclass(frozen=True)
class QueueWorkerArgs:
    """Everything QueueWorker needs from its caller."""

    # Forwarded to provider arguments: accept a literal, an awaitable, or an Output.
    handler: pulumi.Input[str]
    code: pulumi.Input[pulumi.Archive]
    role_arn: Optional[pulumi.Input[str]] = None
    environment: Mapping[str, pulumi.Input[str]] = field(default_factory=dict)

    # Read by this component's own Python logic, so they must be plain values.
    runtime: str = "python3.12"
    batch_size: int = 10
    max_receive_count: int = 5
    timeout_seconds: int = 30
    visibility_timeout_seconds: int = 180
    memory_size: int = 512
    log_retention_days: int = 30
    tags: Mapping[str, str] = field(default_factory=dict)

max_receive_count is a plain int because it is embedded in a redrive policy document the component builds itself. role_arn is Optional[Input[str]] because the caller may either supply an existing role or let the component create one, and that choice is made by an is None test — which is legal on an Optional wrapper but not on the value inside an Output.

2. Freeze the argument object and validate before anything is registered

frozen=True buys two things. It makes the args object hashable and safe to share between several instantiations, and it stops a component from quietly rewriting its own inputs — a habit that makes previews impossible to reason about, because the recorded inputs no longer match what the caller wrote. Mutating one raises dataclasses.FrozenInstanceError: cannot assign to field 'batch_size', and mypy catches the same line statically with error: Property "batch_size" defined in "QueueWorkerArgs" is read-only [misc].

Validation belongs in __post_init__ for single-field rules and in a guard called before super().__init__() for anything that needs the resource name. The ordering is not cosmetic: raising after super().__init__() has already registered the component node leaves the engine with a parent that never receives register_outputs, so the failure is reported as an update error on a resource that half-exists rather than as a clean ValueError from your own code.

# components/queue_worker/args.py (continued)
    def __post_init__(self) -> None:
        if not 1 <= self.batch_size <= 10:
            raise ValueError(f"batch_size must be 1-10 for a standard queue, got {self.batch_size}")
        if not 1 <= self.max_receive_count <= 1000:
            raise ValueError(f"max_receive_count must be 1-1000, got {self.max_receive_count}")
        if not 1 <= self.timeout_seconds <= 900:
            raise ValueError(f"timeout_seconds must be 1-900, got {self.timeout_seconds}")
        if self.visibility_timeout_seconds < self.timeout_seconds * 6:
            raise ValueError(
                "visibility_timeout_seconds must be at least 6x timeout_seconds "
                f"({self.timeout_seconds * 6}), got {self.visibility_timeout_seconds}"
            )
        if not 128 <= self.memory_size <= 10240:
            raise ValueError(f"memory_size must be 128-10240 MB, got {self.memory_size}")

The visibility-timeout rule is the one worth having. AWS will accept a queue whose visibility timeout is shorter than the function timeout, and the symptom appears weeks later as duplicate processing under load. A guard in the argument object turns that into a stack trace on the developer's laptop.

3. Declare the outputs as typed attributes, assign them, then register

Class-level annotations tell readers and mypy exactly what the component publishes. They do not create the attributes — assignment does — so treat the annotation block as a checklist that __init__ must satisfy before it calls register_outputs.

# components/queue_worker/component.py
# CLI: pulumi up --stack dev
from typing import Optional

import pulumi
import pulumi_aws as aws

from .args import QueueWorkerArgs


class QueueWorker(pulumi.ComponentResource):
    """An SQS-backed Lambda worker with a dead-letter queue."""

    queue_url: pulumi.Output[str]
    queue_arn: pulumi.Output[str]
    dead_letter_queue_arn: pulumi.Output[str]
    function_name: pulumi.Output[str]
    log_group_name: pulumi.Output[str]

    def __init__(
        self,
        name: str,
        args: QueueWorkerArgs,
        opts: Optional[pulumi.ResourceOptions] = None,
    ) -> None:
        if not name.islower():
            raise ValueError(f"component name must be lowercase, got {name!r}")
        # State implication: nothing is registered until this line, so a bad arg leaves no node.
        super().__init__("myorg:messaging:QueueWorker", name, None, opts)
        child = pulumi.ResourceOptions(parent=self)

        dlq = aws.sqs.Queue(f"{name}-dlq", tags=dict(args.tags), opts=child)
        queue = aws.sqs.Queue(
            f"{name}-queue",
            visibility_timeout_seconds=args.visibility_timeout_seconds,
            # Provider note: json_dumps keeps dlq.arn an Output, so the DLQ is created first.
            redrive_policy=pulumi.Output.json_dumps(
                {"deadLetterTargetArn": dlq.arn, "maxReceiveCount": args.max_receive_count}
            ),
            tags=dict(args.tags),
            opts=child,
        )

        role_arn: pulumi.Input[str]
        if args.role_arn is None:
            role = aws.iam.Role(
                f"{name}-role",
                assume_role_policy=pulumi.Output.json_dumps(
                    {
                        "Version": "2012-10-17",
                        "Statement": [
                            {
                                "Effect": "Allow",
                                "Principal": {"Service": "lambda.amazonaws.com"},
                                "Action": "sts:AssumeRole",
                            }
                        ],
                    }
                ),
                opts=child,
            )
            aws.iam.RolePolicyAttachment(
                f"{name}-sqs-exec",
                role=role.name,
                policy_arn="arn:aws:iam::aws:policy/service-role/AWSLambdaSQSQueueExecutionRole",
                opts=child,
            )
            role_arn = role.arn
        else:
            role_arn = args.role_arn

        fn = aws.lambda_.Function(
            f"{name}-fn",
            runtime=args.runtime,
            handler=args.handler,
            code=args.code,
            role=role_arn,
            timeout=args.timeout_seconds,
            memory_size=args.memory_size,
            environment=aws.lambda_.FunctionEnvironmentArgs(variables=dict(args.environment)),
            opts=child,
        )
        log_group = aws.cloudwatch.LogGroup(
            f"{name}-logs",
            name=fn.name.apply(lambda n: f"/aws/lambda/{n}"),
            retention_in_days=args.log_retention_days,
            opts=child,
        )
        aws.lambda_.EventSourceMapping(
            f"{name}-esm",
            event_source_arn=queue.arn,
            function_name=fn.arn,
            batch_size=args.batch_size,
            opts=child,
        )

        self.queue_url = queue.url
        self.queue_arn = queue.arn
        self.dead_letter_queue_arn = dlq.arn
        self.function_name = fn.name
        self.log_group_name = log_group.name
        # State implication: register_outputs closes the node and records the published surface.
        self.register_outputs(
            {
                "queue_url": self.queue_url,
                "queue_arn": self.queue_arn,
                "dead_letter_queue_arn": self.dead_letter_queue_arn,
                "function_name": self.function_name,
                "log_group_name": self.log_group_name,
            }
        )

Note self.function_name = fn.name rather than self.function_name = f"{name}-fn". The tempting shortcut is wrong twice. Pulumi auto-names physical resources with a random suffix, so the string does not match reality; more seriously, a plain str carries no dependency information. A consumer that passes it to an aws.cloudwatch.EventTarget gets no edge in the graph, and the engine is free to create the target first — which fails with ResourceNotFoundException: Function not found. Handing back the Output is what makes the ordering correct, and it is the same reasoning behind handling stack outputs and cross-stack references.

4. Configure mypy so the annotations are actually enforced

Annotations are documentation until a checker runs. Both pulumi and generated provider SDKs ship a py.typed marker, so mypy reads their real signatures instead of treating them as Any.

# pyproject.toml
# CLI: mypy
[tool.mypy]
python_version = "3.11"
files = ["components", "__main__.py", "tests"]
strict = true
warn_unreachable = true
show_error_codes = true

[[tool.mypy.overrides]]
module = ["pulumi_aws.*"]
ignore_missing_imports = false

Verification

Type-check first, then preview. A caller mistake now surfaces as a file-and-line error instead of a provider rejection.

# CLI: the contract is enforced before any provider call
mypy
# __main__.py:14: error: Argument "batch_size" to "QueueWorkerArgs" has incompatible type
#   "str"; expected "int"  [arg-type]
# __main__.py:19: error: Argument "handler" to "QueueWorkerArgs" has incompatible type
#   "int"; expected "Union[str, Awaitable[str], Output[str]]"  [arg-type]
pulumi preview --stack dev --diff
pulumi stack output queueUrl --stack dev

If mypy reports error: Skipping analyzing "pulumi_aws": module is installed, but missing library stubs or py.typed marker [import-untyped], it is running from a different environment than the one holding the SDKs; point it at the project virtual environment rather than silencing the code.

Gotchas & Edge Cases

Validation cannot see inside an Output. A guard such as if args.handler.startswith("app.") raises AttributeError: 'Output' object has no attribute 'startswith' the moment a caller passes a resource-derived value. Move that rule into args.handler.apply(_check_handler) and accept that the failure now surfaces during pulumi up, not at construction.

Declared attributes that are never assigned still type-check. A class-level queue_arn: pulumi.Output[str] satisfies mypy even if __init__ forgets the assignment; consumers then hit AttributeError: 'QueueWorker' object has no attribute 'queue_arn' at runtime. Keep the annotation block and the assignment block adjacent and in the same order.

Never default opts to a constructed object. opts: pulumi.ResourceOptions = pulumi.ResourceOptions() is a shared mutable default; every instantiation mutates the same options object. Use Optional[pulumi.ResourceOptions] = None.

Mapping beats dict in the signature. Declaring tags: Mapping[str, str] tells callers the component will not mutate their dictionary, and passing dict(args.tags) into each resource makes that true. Mapping is also covariant in its value type, so a caller holding a dict[str, str] satisfies it without a cast.

A list of Output values is not an Output of a list. list[pulumi.Output[str]] cannot be handed to a provider argument that expects Input[Sequence[Input[str]]] in every position; collapse it with pulumi.Output.all(*items), which returns Output[list[str]] and preserves the dependency on all of them at once.

Adding a required field is a breaking change. Because arguments are positional-capable on a dataclass, inserting a field mid-list also reorders existing call sites. Append new fields with defaults, and reserve removals for a major version of the package described in packaging components for reuse.

Operational Notes

Keep args.py free of provider imports so the argument object can be imported by tests and tooling without loading a multi-megabyte SDK. Ship a py.typed marker inside your own component package, otherwise downstream consumers get Any for every attribute you worked to annotate and the contract stops being enforced at exactly the boundary that matters.

Run mypy in the same CI job that runs pulumi preview, and treat a new error the way you would treat a failed preview. The payoff compounds as the number of stacks instantiating the component grows: a rename of queue_arn becomes a list of files to fix rather than a deployment that half-succeeds. The broader conventions live in Python typing for cloud resource definitions.

FAQ

Should component arguments be a dataclass or a Pydantic model?

A frozen dataclass is enough when the caller is Python and the checker is mypy, and it adds no runtime dependency. Reach for Pydantic when arguments arrive from JSON or YAML and you need coercion and structured error messages rather than a single ValueError.

Why does mypy print Union[str, Awaitable[str], Output[str]] instead of Input[str]?

Input is a plain type alias, not a class, so mypy expands it when reporting. The expansion is a useful reminder of exactly what a caller is permitted to pass.

Do I need Output annotations if nobody on the team runs mypy?

They still pay for themselves through editor autocomplete and as a machine-checkable list of what register_outputs must publish. Without a checker, though, nothing stops a plain str from being assigned to an attribute declared as Output[str].

Can I validate an argument that arrives as an Output?

Only inside apply, because the value does not exist until the producing resource is created. Validate plain arguments eagerly in __post_init__ and keep apply-time checks to the few rules that genuinely need a resolved value.

Does every public attribute have to appear in register_outputs?

No — attributes are readable from the object regardless. Registering them records the values against the component's URN so they appear in state and in pulumi stack export, which is what you want for anything another stack or a human operator will look up.

Related

• Pulumi Component Resources — the parent topic covering parenting, registration, and component structure.
• Building a Reusable VPC Component in Pulumi (Python) — the end-to-end authoring walkthrough this typing discipline applies to.
• Testing Pulumi Component Resources in Isolation — assert that the declared surface is actually produced.
• Python Typing for Cloud Resource Definitions — the site-wide typing conventions these components follow.