Skip to content

Resources API

Base classes and utilities for creating Terraform resources with full CRUD lifecycle management.

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

Overview

Resources in Pyvider represent manageable infrastructure components that Terraform can plan and apply through Pyvider's async lifecycle.

Key Components

  • BaseResource - Base class for all resources
  • @register_resource - Decorator for resource registration
  • Resource Context - Per-operation context with provider access
  • Private State - Encrypted storage for sensitive data
  • Lifecycle Protocols - Standard CRUD interfaces

Lifecycle Methods

Resources interact with Terraform via a plan/apply cycle: - read(ctx: ResourceContext) — refresh the latest state (called by Terraform refresh and after apply) - plan(ctx) — framework-provided method that calls _create/_update/_delete_plan hooks to build a planned state - apply(ctx) — framework-provided method that calls _create_apply/_update_apply/_delete_apply hooks to enact the plan

Resource authors typically override: - _create(ctx, base_plan) / _update(ctx, base_plan) / _delete_plan(ctx) to shape the plan output - _create_apply(ctx) / _update_apply(ctx) / _delete_apply(ctx) to perform real API calls and return final state/private state tuples

See src/pyvider/resources/base.py for the exact signatures.

Usage Examples

Basic Resource Implementation

Python
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 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
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
import attrs
from pyvider.resources import register_resource, BaseResource
from pyvider.resources.context import ResourceContext
from pyvider.schema import s_resource, a_str, a_num, PvsSchema


@attrs.define
class ServerConfig:
    """Resource configuration."""
    name: str
    size: str = "medium"


@attrs.define
class ServerState:
    """Resource state."""
    id: str
    name: str
    size: str
    status: str


@register_resource("server")
class Server(BaseResource):
    """Manages a server resource."""

    config_class = ServerConfig
    state_class = ServerState

    @classmethod
    def get_schema(cls) -> PvsSchema:
        return s_resource({
            # Config attributes
            "name": a_str(required=True, description="Server name"),
            "size": a_str(default="medium", description="Server size"),

            # Computed attributes
            "id": a_str(computed=True, description="Server ID"),
            "status": a_str(computed=True, description="Server status"),
        })

    async def _create_apply(
        self,
        ctx: ResourceContext,
    ) -> tuple[ServerState | None, None]:
        """Create a server."""
        # Get provider instance
        from pyvider.hub import hub
        provider = hub.get_component("singleton", "provider")

        # Create server via API
        server = await provider.api.create_server(
            name=ctx.config.name,
            size=ctx.config.size,
        )

        # Return state
        return ServerState(
            id=server.id,
            name=server.name,
            size=server.size,
            status=server.status,
        ), None

    async def read(self, ctx: ResourceContext) -> ServerState | None:
        """Read current server state."""
        if not ctx.state:
            return None

        from pyvider.hub import hub
        provider = hub.get_component("singleton", "provider")

        try:
            server = await provider.api.get_server(ctx.state.id)
            return ServerState(
                id=server.id,
                name=server.name,
                size=server.size,
                status=server.status,
            )
        except ResourceNotFoundError:
            return None  # Server was deleted

    async def _update_apply(
        self,
        ctx: ResourceContext,
    ) -> tuple[ServerState | None, None]:
        """Update a server."""
        from pyvider.hub import hub
        provider = hub.get_component("singleton", "provider")

        server = await provider.api.update_server(
            server_id=ctx.state.id,
            name=ctx.config.name,
            size=ctx.config.size,
        )

        return ServerState(
            id=server.id,
            name=server.name,
            size=server.size,
            status=server.status,
        ), None

    async def _delete_apply(self, ctx: ResourceContext) -> None:
        """Delete a server."""
        from pyvider.hub import hub
        provider = hub.get_component("singleton", "provider")

        await provider.api.delete_server(ctx.state.id)

Resource with Private State

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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
54
55
56
57
@register_resource("database")
class Database(BaseResource):
    """Database with encrypted credentials in private state."""

    async def _create_apply(
        self,
        ctx: ResourceContext,
    ) -> tuple[State | None, dict | None]:
        """Create database with credentials."""
        from pyvider.hub import hub
        provider = hub.get_component("singleton", "provider")

        db = await provider.api.create_database(ctx.config)

        # Public state (visible in terraform.tfstate)
        state = State(
            id=db.id,
            endpoint=db.endpoint,
            port=db.port,
        )

        # Private state (encrypted, not in terraform.tfstate)
        private = {
            "master_password": db.master_password,
            "admin_token": db.admin_token,
        }

        return state, private

    async def _update_apply(
        self,
        ctx: ResourceContext,
    ) -> tuple[State | None, dict | None]:
        """Update with access to private state."""
        # Access private state
        current_password = ctx.private_state.get("master_password") if ctx.private_state else None

        # Update logic here
        from pyvider.hub import hub
        provider = hub.get_component("singleton", "provider")

        db = await provider.api.update_database(
            db_id=ctx.state.id,
            config=ctx.config,
        )

        # Return updated state and private state
        state = State(
            id=db.id,
            endpoint=db.endpoint,
            port=db.port,
        )

        # Keep or update private state
        private = ctx.private_state or {}

        return state, private

Resource with Validation

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
@register_resource("validated_resource")
class ValidatedResource(BaseResource):
    """Resource with custom validation."""

    @classmethod
    def get_schema(cls) -> PvsSchema:
        return s_resource({
            "name": a_str(
                required=True,
                validators=[
                    lambda x: len(x) >= 3 or "Name too short",
                    lambda x: len(x) <= 64 or "Name too long",
                    lambda x: x.isalnum() or "Name must be alphanumeric",
                ],
            ),
            "port": a_num(
                required=True,
                validators=[
                    lambda x: 1 <= x <= 65535 or "Invalid port number",
                ],
            ),
        })

Resource with Import Support

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@register_resource("importable_resource")
class ImportableResource(BaseResource):
    """Resource that supports Terraform import."""

    async def import_resource(
        self,
        import_id: str,
    ) -> tuple[dict, None]:
        """Import existing resource by ID."""
        from pyvider.hub import hub
        provider = hub.get_component("singleton", "provider")

        # Fetch existing resource
        resource = await provider.api.get_resource(import_id)

        if not resource:
            raise ResourceNotFoundError(f"Resource {import_id} not found")

        # Return state as dict
        return {
            "id": resource.id,
            "name": resource.name,
            "status": resource.status,
        }, None

Testing Resources

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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
import pytest
from pyvider.resources.context import ResourceContext


@pytest.mark.asyncio
async def test_server_create():
    """Test server creation."""
    server = Server()

    ctx = ResourceContext(
        config=ServerConfig(name="test-server", size="large")
    )

    state, private = await server._create_apply(ctx)

    assert state is not None
    assert state.name == "test-server"
    assert state.size == "large"
    assert state.id


@pytest.mark.asyncio
async def test_server_lifecycle():
    """Test full server lifecycle."""
    server = Server()

    # Create
    create_ctx = ResourceContext(
        config=ServerConfig(name="test-server")
    )
    state, _ = await server._create_apply(create_ctx)
    assert state.status == "running"

    # Read
    read_ctx = ResourceContext(state=state)
    read_state = await server.read(read_ctx)
    assert read_state.id == state.id

    # Update
    update_ctx = ResourceContext(
        config=ServerConfig(name="test-server", size="xlarge"),
        state=state,
    )
    updated_state, _ = await server._update_apply(update_ctx)
    assert updated_state.size == "xlarge"

    # Delete
    delete_ctx = ResourceContext(state=updated_state)
    await server._delete_apply(delete_ctx)

Module Reference