Lifecycle Methods

Every resource implements lifecycle methods that Pragmatiks calls when managing your resources. Handlers are observe-then-act: each resource owns exactly one external object, findable from its identity and config alone. on_observe reports whether that object exists and what it looks like; on_create, on_update, and on_delete converge it to the configuration.

Method Signatures

from pragma_sdk import Config, Outputs, Resource


class MyResource(Resource[MyConfig, MyOutputs]):
    """Resource that owns one external object."""

    async def on_create(self) -> MyOutputs:
        """Called when creating a new resource."""
        ...

    async def on_observe(self) -> MyOutputs | None:
        """Called to report the external object, or None when it does not exist."""
        ...

    async def on_update(self, previous_config: MyConfig | None) -> MyOutputs:
        """Called when updating an existing resource."""
        ...

    async def on_delete(self) -> None:
        """Called when deleting a resource."""
        ...

All lifecycle methods are async and run within the provider runtime. Do I/O through an async client or await asyncio.to_thread(...), so a handler yields to the event loop and can be cancelled. on_observe is mandatory unless the resource is computed. Registering a non-computed class without it raises MissingObserveError when the provider package is imported, so pragma providers publish fails naming the resource type.

Instance Attributes

Inside lifecycle methods, you have access to several instance attributes:

Attribute	Type	Description
`self.name`	`str`	Resource instance name
`self.project_id`	`str`	Project the resource belongs to
`self.config`	`ConfigT`	Current configuration, with dependencies resolved
`self.outputs`	`OutputsT \| None`	Never populated when a handler runs. Read state from the external system instead
`self.tags`	`list[str] \| None`	Tags attached to the resource

on_create

Called when a resource is first created. Your implementation should provision the underlying infrastructure and return the outputs.

async def on_create(self) -> DatabaseOutputs:
    database = await self.client.create_database(
        name=self.config.name,
        size=self.config.size_gb,
    )

    return DatabaseOutputs(
        connection_url=database.connection_string,
        database_id=database.id,
    )

When called: The first time a resource is applied. Return value: An instance of your Outputs class containing values that other resources can reference.

on_observe

Reports the external object this resource owns. Locate it solely from identity (self.project_id, self.name) and self.config, including resolved dependencies, never from self.outputs. Read only the external system, never other resources; a handler that needs another resource declares a Dependency.

async def on_observe(self) -> DatabaseOutputs | None:
    try:
        database = await self.client.get_database(self.config.name)
    except NotFoundError:
        return None

    return DatabaseOutputs(
        connection_url=database.connection_string,
        database_id=database.id,
    )

Return value: None when no object exists for this identity. Otherwise the same Outputs the write path returned for that state. Errors: Let anything other than “not found” propagate. An exception means the observation failed, not that the object is absent. When the external system can hold more than one match for the key, raise on more than one match. Stable outputs: Every output field is derived from identity or config, or read from the object once the write path has waited for it to be ready, and on_observe reads it the same way. Values that change without a platform write, such as replica readiness, counters, or indexing status, belong in health(), never in outputs.

on_update

Called to converge an existing object to the current configuration. Receives the previous configuration so you can limit the work to what changed.

async def on_update(self, previous_config: DatabaseConfig | None) -> DatabaseOutputs:
    database = await self.client.get_database(self.config.name)

    if previous_config is None or previous_config.size_gb != self.config.size_gb:
        database = await self.client.resize_database(
            database_id=database.id,
            new_size=self.config.size_gb,
        )

    return DatabaseOutputs(
        connection_url=database.connection_string,
        database_id=database.id,
    )

When called: Each time the resource configuration changes after initial creation, and when a create finds the object already present. Parameters:

previous_config: The configuration from the previous successful apply, or None when the object exists but the platform holds no prior configuration. On None, skip “unchanged, return early” shortcuts and apply the full desired state.

Return value: An instance of your Outputs class, read from the converged object.

Common Update Patterns

Immutable fields - Fields that cannot change in place, and every field on_observe reads to locate the object, are declared ImmutableField or ImmutableDependency. The platform refuses a change to them, so a user changes one by deleting and recreating the resource. The handler never checks them:

class SecretConfig(Config):
    """Configuration for a secret.

    Attributes:
        project_id: Project holding the secret. Locates the secret, so it cannot change.
        secret_id: Secret name. Locates the secret, so it cannot change.
        data: Secret payload.
    """

    project_id: ImmutableField[str]
    secret_id: ImmutableField[str]
    data: SensitiveField[str]

No-op when unchanged - Skip work when the live object already matches the configuration:

async def on_update(self, previous_config: SecretConfig | None) -> SecretOutputs:
    latest = await self.client.get_latest_version(self.config.secret_id)

    if latest.data != self.config.data:
        latest = await self.client.add_version(data=self.config.data)

    return SecretOutputs(version_id=latest.id)

on_delete

Called when a resource is being removed. Clean up the underlying infrastructure.

async def on_delete(self) -> None:
    try:
        await self.client.delete_database(self.config.name)
    except NotFoundError:
        return

When called: When a resource is removed from a graph or explicitly deleted. Return value: None. Any return value is ignored.

Idempotency Requirements

All lifecycle methods must be idempotent. The same method may be called multiple times with the same input, and must produce the same result. Creating an object that already exists converges it through on_update(previous_config=None); deleting an object that is already gone succeeds.

This happens when:

The runtime restarts after executing a method but before confirming completion
Network issues cause retries
The user re-applies the same configuration

Idempotent Create

Handle the case where a previous attempt already created the resource by reading it instead:

async def on_create(self) -> SecretOutputs:
    try:
        secret = await self.client.create_secret(self.config.secret_id)
    except AlreadyExistsError:
        secret = await self.client.get_secret(self.config.secret_id)

    return SecretOutputs(resource_name=secret.name)

Idempotent Delete

Handle the case where the resource is already gone; that counts as success:

async def on_delete(self) -> None:
    try:
        await self.client.delete_secret(self.config.secret_id)
    except NotFoundError:
        return

Idempotent Update

Compare the live object with the configuration and write only the difference. With previous_config=None, never assume nothing changed:

async def on_update(self, previous_config: MyConfig | None) -> MyOutputs:
    current = await self.client.get_object(self.config.name)

    if current.settings != self.config.settings:
        current = await self.client.update_object(self.config.name, self.config.settings)

    return MyOutputs(object_id=current.id)

Error Handling

Raise exceptions to signal failures. The runtime will:

Mark the resource as failed
Record the error message
Stop processing dependent resources

async def on_create(self) -> DatabaseOutputs:
    try:
        database = await self.client.create_database(self.config.name)
    except QuotaExceededError as error:
        raise RuntimeError(f"Quota exceeded: {error}") from error
    except InvalidConfigError as error:
        raise ValueError(f"Invalid configuration: {error}") from error

    return DatabaseOutputs(connection_url=database.url)

Use specific exception types:

ValueError - Invalid configuration (user error)
RuntimeError - Infrastructure or API failures
Let unexpected exceptions propagate with full tracebacks

Complete Example

from mycloud import CloudClient, CloudDatabase
from mycloud.exceptions import AlreadyExists, NotFound
from pragma_sdk import Config, Field, ImmutableField, Outputs, Provider, Resource


class DatabaseConfig(Config):
    """Configuration for a managed database.

    Attributes:
        name: Database name. Locates the database, so it cannot change.
        size_gb: Storage size in gigabytes.
        region: Region hosting the database. Cannot change in place.
    """

    name: ImmutableField[str]
    size_gb: Field[int] = 10
    region: ImmutableField[str] = "us-east-1"


class DatabaseOutputs(Outputs):
    """Outputs of a managed database.

    Attributes:
        database_id: Identifier assigned by the cloud.
        connection_url: Connection string for clients.
        host: Database host name.
    """

    database_id: str
    connection_url: str
    host: str


mycloud = Provider()


@mycloud.resource("database")
class Database(Resource[DatabaseConfig, DatabaseOutputs]):
    """Managed database located by its name and region."""

    def build_client(self) -> CloudClient:
        """Build an async client for the configured region.

        Returns:
            Cloud API client.
        """
        return CloudClient(region=self.config.region)

    def build_outputs(self, database: CloudDatabase) -> DatabaseOutputs:
        """Build outputs from a database read from the cloud.

        Args:
            database: Database as the cloud API reports it.

        Returns:
            Outputs for the database.
        """
        return DatabaseOutputs(
            database_id=database.id,
            connection_url=database.connection_string,
            host=database.host,
        )

    async def on_create(self) -> DatabaseOutputs:
        """Create the database, or adopt it when a previous attempt created it.

        Returns:
            Outputs for the database.
        """
        client = self.build_client()

        try:
            database = await client.create_database(
                name=self.config.name,
                size_gb=self.config.size_gb,
            )
        except AlreadyExists:
            database = await client.get_database(self.config.name)

        return self.build_outputs(database)

    async def on_observe(self) -> DatabaseOutputs | None:
        """Report the database named by config.

        Returns:
            Outputs for the database, or ``None`` when it does not exist.
        """
        try:
            database = await self.build_client().get_database(self.config.name)
        except NotFound:
            return None

        return self.build_outputs(database)

    async def on_update(self, previous_config: DatabaseConfig | None) -> DatabaseOutputs:
        """Resize the database when its live size differs from config.

        Args:
            previous_config: Configuration last applied, or ``None``. Unused,
                because the live size is compared with config directly.

        Returns:
            Outputs for the converged database.
        """
        client = self.build_client()
        database = await client.get_database(self.config.name)

        if database.size_gb != self.config.size_gb:
            database = await client.resize_database(
                database_id=database.id,
                new_size_gb=self.config.size_gb,
            )

        return self.build_outputs(database)

    async def on_delete(self) -> None:
        """Delete the database.

        Succeeds when the database is already gone.
        """
        try:
            await self.build_client().delete_database(self.config.name)
        except NotFound:
            return

Computed Resources

A resource with no external object of its own declares computed = True. It needs no on_observe: its outputs derive from its configuration, or from the subresources it owns. Composition resources that only assemble subresources are computed.

@mycloud.resource("connection")
class Connection(Resource[ConnectionConfig, ConnectionOutputs]):
    """Connection settings derived from configuration alone."""

    computed = True

    async def on_create(self) -> ConnectionOutputs:
        """Build the connection URL from config.

        Returns:
            Outputs holding the URL.
        """
        return ConnectionOutputs(url=f"postgres://{self.config.host}:{self.config.port}")

    async def on_update(self, previous_config: ConnectionConfig | None) -> ConnectionOutputs:
        """Rebuild the connection URL from config.

        Args:
            previous_config: Configuration last applied, or ``None``.

        Returns:
            Outputs holding the URL.
        """
        return await self.on_create()

    async def on_delete(self) -> None:
        """Nothing to delete: a computed resource owns no external object."""

Creating Subresources

Providers can create and manage subresources from within lifecycle handlers. This enables composition patterns where a high-level resource provisions multiple lower-level resources.

The Pattern

Build the subresource instance, set this resource as its owner, and apply it:

async def on_create(self) -> AppOutputs:
    database = Database(
        project_id=self.project_id,
        name=f"{self.name}-db",
        config=DatabaseConfig(name=f"{self.name}-db", size_gb=10),
    )
    database.set_owner(self)
    await database.apply()
    return AppOutputs(database_name=database.name)

The handler never waits for the subresource. apply() returns once the platform accepts the subresource, and the platform provisions it after the handler returns. A resource that needs the subresource’s outputs, such as its connection string, declares a Dependency or a field reference on it; the platform runs that resource only once the subresource is ready, with the values resolved. See Configuration and Outputs.

Methods

set_owner(parent) - Establishes an ownership relationship. When the parent resource is deleted, owned subresources are automatically cleaned up via cascading deletes. Returns self for method chaining. apply() - Sends the resource to the API for creation or update and returns once the API accepts it. The platform then provisions the resource with the provider that handles its type. Returns self for method chaining.

Never wait inside a handler for another resource to become ready: no wait_ready, no apply(wait=True), no sleep or retry loop. Declare the dependency and let the platform sequence the work.

Complete Example

Here’s a complete example of an application resource that provisions its own database, using the Database resource from the complete example above. The application owns no external object of its own, so it is computed, and its outputs derive from its own identity:

from pragma_sdk import Config, Field, Outputs, Resource


class AppConfig(Config):
    """Configuration for an application environment.

    Attributes:
        database_size_gb: Storage size of the application's database in gigabytes.
    """

    database_size_gb: Field[int] = 10


class AppOutputs(Outputs):
    """Outputs of an application environment.

    Attributes:
        database_name: Name of the database the application owns.
    """

    database_name: str


@mycloud.resource("app")
class App(Resource[AppConfig, AppOutputs]):
    """Application environment that owns its database."""

    computed = True

    async def on_create(self) -> AppOutputs:
        """Apply the owned database.

        Returns:
            Outputs holding the database name.
        """
        database = Database(
            project_id=self.project_id,
            name=f"{self.name}-db",
            config=DatabaseConfig(name=f"{self.name}-db", size_gb=self.config.database_size_gb),
        )
        database.set_owner(self)
        await database.apply()
        return AppOutputs(database_name=database.name)

    async def on_update(self, previous_config: AppConfig | None) -> AppOutputs:
        """Re-apply the database with the current size.

        Args:
            previous_config: Configuration last applied, or ``None``.

        Returns:
            Outputs holding the database name.
        """
        return await self.on_create()

    async def on_delete(self) -> None:
        """Nothing to delete: owned subresources are deleted by cascading deletes."""

When to Use Subresources

Use Case	Approach
Encapsulated infrastructure (DB, cache, etc.)	Subresources with ownership
Shared resources across multiple parents	Dependencies without ownership
User-managed separate resources	Separate top-level resources

Use subresources when the child resource’s lifecycle is tightly coupled to the parent. If users might want to manage the child independently, expose it as a separate resource type instead.

Getting Started

Concepts

Resource Catalog

Agno (AI Agents)

GCP

Kubernetes

Building Providers

Method Signatures

Instance Attributes

on_create

on_observe

on_update

Common Update Patterns

on_delete

Idempotency Requirements

Idempotent Create

Idempotent Delete

Idempotent Update

Error Handling

Complete Example

Computed Resources

Creating Subresources

The Pattern

Methods

Complete Example

When to Use Subresources

​Method Signatures

​Instance Attributes

​on_create

​on_observe

​on_update

​Common Update Patterns

​on_delete

​Idempotency Requirements

​Idempotent Create

​Idempotent Delete

​Idempotent Update

​Error Handling

​Complete Example

​Computed Resources

​Creating Subresources

​The Pattern

​Methods

​Complete Example

​When to Use Subresources

Method Signatures

Instance Attributes

on_create

on_observe

on_update

Common Update Patterns

on_delete

Idempotency Requirements

Idempotent Create

Idempotent Delete

Idempotent Update

Error Handling

Complete Example

Computed Resources

Creating Subresources

The Pattern

Methods

Complete Example

When to Use Subresources