Writing a Pulumi Dynamic Provider in Python

This guide implements a complete dynamic.ResourceProvider in Python that manages a resource against an external REST API — covering create, read, update, delete, and diff, with the CreateResult and DiffResult return types and the serialization caveats that trip people up. It is the hands-on counterpart to Dynamic Providers and Custom Resources in Pulumi, itself part of the broader Pulumi Patterns & Provider Management workflow.

Context

When an API has no Pulumi provider, a dynamic provider gives it managed, idempotent lifecycle handling instead of fire-and-forget imperative calls. The example below manages a "widget" on a fictional api.example.com service: a thing with a name and a size that the API creates, returns by id, updates in place, and deletes. The same shape applies to any CRUD-capable REST or SDK endpoint.

Prerequisites

  • Python 3.9+ with pulumi>=3.0 and requests: pip install pulumi requests.
  • An initialized Pulumi project and stack — verify with pulumi stack ls.
  • The external API's base URL and a token, supplied through Pulumi config: pulumi config set --secret apiToken <token>.
  • IAM/account permission on the external service to create, read, update, and delete the resource.

Implementation

1. Define typed inputs and the resource client

Keep the API interaction in small functions so the lifecycle methods stay readable. Use a dataclass for the user-facing inputs.

# providers/widget.py — used by __main__.py; run with: pulumi up
from dataclasses import dataclass

@dataclass
class WidgetArgs:
    name: str
    size: int  # Provider note: these map one-to-one to the external API's JSON body.

API_BASE = "https://api.example.com/widgets"

2. Implement the ResourceProvider with full CRUD and diff

Every method imports requests locally. The provider class is serialized into state, so it must not capture a module-level client, a connection, or any unpicklable object.

# providers/widget.py — pulumi up
from typing import Optional
from pulumi.dynamic import (
    ResourceProvider, CreateResult, ReadResult, UpdateResult, DiffResult,
)

class WidgetProvider(ResourceProvider):
    def _headers(self, props: dict) -> dict:
        # Provider note: token is passed in via props, not captured at class scope.
        return {"Authorization": f"Bearer {props['api_token']}"}

    def create(self, props: dict) -> CreateResult:
        import requests  # serialization caveat: import inside the method.
        r = requests.post(
            API_BASE,
            json={"name": props["name"], "size": props["size"]},
            headers=self._headers(props), timeout=10,
        )
        r.raise_for_status()
        rid = r.json()["id"]
        # State implication: id_ is permanent; outs are this resource's stored state.
        return CreateResult(id_=rid, outs={**props, "remote_id": rid})

    def read(self, id_: str, props: dict) -> ReadResult:
        import requests
        r = requests.get(f"{API_BASE}/{id_}", headers=self._headers(props), timeout=10)
        r.raise_for_status()
        body = r.json()
        # State implication: reconciles checkpoint with live state on `pulumi refresh`.
        return ReadResult(id_=id_, outs={**props, "name": body["name"], "size": body["size"]})

    def diff(self, id_: str, old: dict, new: dict) -> DiffResult:
        fields = ("name", "size")
        changes = [f for f in fields if old.get(f) != new.get(f)]
        # name change forces replacement; size can update in place.
        return DiffResult(changes=bool(changes), replaces=["name"] if "name" in changes else [])

    def update(self, id_: str, old: dict, new: dict) -> UpdateResult:
        import requests
        r = requests.put(
            f"{API_BASE}/{id_}",
            json={"size": new["size"]},
            headers=self._headers(new), timeout=10,
        )
        r.raise_for_status()
        return UpdateResult(outs={**new, "remote_id": id_})

    def delete(self, id_: str, props: dict) -> None:
        import requests
        # State implication: called on resource removal and on replacement teardown.
        requests.delete(f"{API_BASE}/{id_}", headers=self._headers(props), timeout=10).raise_for_status()

3. Expose a typed Resource wrapper

Wrap the provider so consumers get a clean class with typed outputs instead of raw dicts. Feed the secret token from Pulumi config.

# __main__.py — pulumi up
import pulumi
from pulumi.dynamic import Resource
from providers.widget import WidgetProvider, WidgetArgs

class Widget(Resource):
    remote_id: pulumi.Output[str]

    def __init__(self, name: str, args: WidgetArgs, opts: Optional[pulumi.ResourceOptions] = None) -> None:
        token = pulumi.Config().require_secret("apiToken")
        props = {**vars(args), "api_token": token, "remote_id": None}
        # State implication: api_token is stored as a secret because it is an Output secret.
        super().__init__(WidgetProvider(), name, props, opts)

widget = Widget("primary", WidgetArgs(name="checkout", size=3))
pulumi.export("widget_id", widget.remote_id)

Verification

Apply twice to prove idempotency, then refresh to prove drift detection works.

# pulumi up
pulumi up --yes            # creates the widget
pulumi up --yes            # expect "no changes": diff converged
pulumi stack output widget_id
pulumi refresh --yes       # read() reconciles any out-of-band change
# python -m pytest test_widget_provider.py
def test_diff_no_change_is_idempotent() -> None:
    p = WidgetProvider()
    state = {"name": "checkout", "size": 3}
    assert p.diff("id-1", state, dict(state)).changes is False

def test_name_change_forces_replacement() -> None:
    p = WidgetProvider()
    res = p.diff("id-1", {"name": "a", "size": 3}, {"name": "b", "size": 3})
    assert res.replaces == ["name"]

Gotchas & Edge Cases

TypeError: cannot pickle '_io.BufferedReader' (or similar) on pulumi up. The provider captured an unpicklable object. Construct clients and import libraries inside each method, and pass credentials through props rather than storing them on the instance — exactly as the example does with api_token.

Secrets leak into plaintext state. A token passed as a normal string is stored unencrypted. Source it from Config().require_secret(...) so it arrives as a secret Output; Pulumi then encrypts it in the checkpoint. Never hardcode the token in WidgetArgs.

Replacement deletes before it creates and breaks dependents. A replaces field triggers delete-then-create by default. If the resource backs something that cannot tolerate a gap, set delete_before_replace=False via ResourceOptions, or design the API call to be updatable so name need not force replacement.