Skip to content

Best Practices for Provider Development

This guide provides foundational best practices for developing Pyvider providers, focusing on design patterns, code organization, and development standards. These patterns are derived from real-world usage and the battle-tested pyvider-components repository.

🤖 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.

For operational concerns like error handling, logging, performance, testing, and security, see the Production Readiness Guide.

Table of Contents

Provider Design Patterns

Single Responsibility Principle

Keep each component focused on a single, well-defined responsibility:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# Good: Focused resource
@register_resource("pyvider_file_content")
class FileContentResource(BaseResource):
    """Manages file content with atomic writes."""
    pass

# Bad: Resource doing too much
@register_resource("pyvider_file_manager")
class FileManager(BaseResource):
    """Manages files, directories, permissions, and backups."""  # Too much!
    pass

Component Selection

Choose the right component type for your use case:

Component Use When Example
Resource Managing infrastructure with lifecycle (CRUD) file_content, local_directory
Data Source Reading existing data env_variables, file_info, http_api
Function Stateless transformations string formatting, jq queries
Ephemeral Short-lived resources (sessions, tokens) timed_token, database_connection

Naming Conventions

Follow consistent naming patterns:

Python
1
2
3
4
5
6
7
8
# Good naming
@register_resource("pyvider_file_content")  # prefix_noun
@register_data_source("pyvider_env_variables")  # prefix_noun_plural
@register_function(name="format_string")  # verb_noun

# Bad naming
@register_resource("pyvider_manage_file")  # don't use verbs for resources
@register_data_source("pyvider_api")  # too generic

Provider Configuration Design

Keep provider configuration simple and focused:

Python
1
2
3
4
5
6
7
8
@define(frozen=True)
class ProviderConfig:
    # Core settings only
    api_endpoint: str = field(default="https://api.example.com")
    api_key: str = field(metadata={"sensitive": True})
    timeout: int = field(default=30)

    # Avoid: Complex nested structures, business logic

Schema Design Best Practices

Use Descriptive Names and Documentation

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
# Good: Clear, documented schema
@classmethod
def get_schema(cls) -> PvsSchema:
    return s_resource({
        "filename": a_str(
            required=True,
            description="Path to the file to manage"
        ),
        "content": a_str(
            required=True,
            description="Content to write to the file"
        ),
        "content_hash": a_str(
            computed=True,
            description="SHA256 hash of the file content"
        ),
        "exists": a_bool(
            computed=True,
            description="Whether the file exists on disk"
        ),
    })

# Bad: No descriptions, unclear names
@classmethod
def get_schema(cls) -> PvsSchema:
    return s_resource({
        "f": a_str(required=True),  # What is 'f'?
        "c": a_str(required=True),  # What is 'c'?
        "h": a_str(computed=True),  # No context
    })

Mark Computed Attributes Correctly

Python
1
2
3
4
5
6
7
# Computed attributes are calculated by the provider
"content_hash": a_str(computed=True)  # Provider calculates
"exists": a_bool(computed=True)  # Provider determines

# Required/optional attributes come from user
"filename": a_str(required=True)  # User must provide
"permissions": a_str(default="644")  # User can override

Use Validators

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from pyvider.schema import a_str, a_num

# Validate inputs to prevent errors
@classmethod
def get_schema(cls) -> PvsSchema:
    return s_resource({
        "port": a_num(
            required=True,
            validators=[
                lambda x: 1 <= x <= 65535 or "Port must be 1-65535"
            ]
        ),
        "protocol": a_str(
            required=True,
            validators=[
                lambda x: x in ["http", "https"] or "Must be http or https"
            ]
        ),
    })

Keep Schemas Simple

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
# Good: Flat, focused schema
{
    "name": a_str(required=True),
    "size": a_num(default=10),
    "enabled": a_bool(default=True),
}

# Avoid: Deeply nested complexity without good reason
{
    "config": a_obj({
        "section1": a_obj({
            "subsection": a_obj({
                "deep_value": a_str()  # Too deep
            })
        })
    })
}

Resource Implementation Patterns

Implement All CRUD Methods

All resources must implement the complete lifecycle:

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
@register_resource("pyvider_example")
class ExampleResource(BaseResource):

    async def read(self, ctx: ResourceContext) -> ExampleState | None:
        """
        Read current state. Return None if resource doesn't exist.
        Called during refresh and before updates.
        """
        if not resource_exists:
            return None
        return ExampleState(...)

    async def _create(self, ctx: ResourceContext, base_plan: dict) -> tuple[dict | None, None]:
        """
        Create new resource. Return (state_dict, None).
        """
        # Create the resource
        result = await self.api.create(...)
        return {**base_plan, "id": result.id}, None

    async def _update(self, ctx: ResourceContext, base_plan: dict) -> tuple[dict | None, None]:
        """
        Update existing resource. Return (state_dict, None).
        """
        # Update the resource
        await self.api.update(ctx.state.id, ...)
        return base_plan, None

    async def _delete(self, ctx: ResourceContext) -> None:
        """
        Delete resource. No return value.
        """
        await self.api.delete(ctx.state.id)

Use Async/Await Consistently

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# Good: All I/O is async
async def read(self, ctx: ResourceContext) -> State | None:
    content = await async_read_file(path)
    result = await self.api_client.get(url)
    return State(...)

# Bad: Blocking I/O
async def read(self, ctx: ResourceContext) -> State | None:
    content = open(path).read()  # Blocks!
    result = requests.get(url)  # Blocks!
    return State(...)

Handle Missing Resources Gracefully

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
async def read(self, ctx: ResourceContext) -> FileContentState | None:
    """Return None if resource doesn't exist."""
    filename = ctx.state.filename if ctx.state else ctx.config.filename
    path = Path(filename)

    # Don't raise errors for missing resources
    if not path.is_file():
        logger.debug("File does not exist", path=str(path))
        return None  # Terraform will handle this

    # Read and return state
    content = safe_read_text(path)
    return FileContentState(...)

Ensure Idempotency

Operations should be safe to run multiple times:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
async def _create(self, ctx: ResourceContext, base_plan: dict) -> tuple[dict | None, None]:
    path = Path(base_plan["filename"])

    # Ensure parent directory exists (idempotent)
    ensure_dir(path.parent)

    # Atomic write (idempotent - overwrites if exists)
    atomic_write_text(path, base_plan["content"])

    # Calculate hash
    content_hash = hashlib.sha256(base_plan["content"].encode()).hexdigest()

    return {**base_plan, "content_hash": content_hash, "exists": True}, None

Type Safety

Use Attrs Classes for Data Models

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from attrs import define, field

@define(frozen=True)  # Immutable
class FileContentConfig:
    """Configuration for file content resource."""
    filename: str = field()
    content: str = field()

    @filename.validator
    def _validate_filename(self, attribute, value):
        if not value:
            raise ValueError("filename cannot be empty")

@define(frozen=True)
class FileContentState:
    """State for file content resource."""
    filename: str = field()
    content: str = field()
    exists: bool | None = field(default=None)
    content_hash: str | None = field(default=None)

Leverage Type Hints

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from typing import Any

class MyResource(BaseResource["my_resource", MyState, MyConfig]):
    config_class = MyConfig
    state_class = MyState

    async def read(self, ctx: ResourceContext) -> MyState | None:
        """Type hints help catch errors early."""
        result: MyState | None = await self._fetch_state()
        return result

    async def _create(
        self,
        ctx: ResourceContext,
        base_plan: dict[str, Any]
    ) -> tuple[dict[str, Any] | None, bytes | None]:
        """Clear parameter and return types."""
        pass

Use Type Checkers

Bash
1
2
3
# Run type checking in CI/CD
uv run mypy src/pyvider --strict
uv run pyright src/pyvider

Code Organization

Single Responsibility Per File

Text Only
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
my_provider/
├── resources/
│   ├── __init__.py
│   ├── file_content.py      # One resource
│   └── local_directory.py   # One resource
├── data_sources/
│   ├── __init__.py
│   ├── env_variables.py     # One data source
│   └── file_info.py         # One data source
└── functions/
    ├── __init__.py
    └── string_utils.py      # Related functions

Use Capabilities for Shared Logic

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# Instead of duplicating code across resources
from pyvider.capabilities import requires_capability

@register_resource("my_resource")
@requires_capability("caching")
class MyResource(BaseResource):

    async def read(self, ctx: ResourceContext) -> State | None:
        # Use shared caching capability
        cached = await self.capabilities.caching.get(cache_key)
        if cached:
            return cached

        result = await self._fetch_from_api()
        await self.capabilities.caching.set(cache_key, result)
        return result

Documentation Standards

Write Clear Docstrings

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
@register_resource("pyvider_file_content")
class FileContentResource(BaseResource):
    """
    Manages file content with atomic writes and content tracking.

    This resource creates and manages text files on the local filesystem.
    It provides:
    - Atomic write operations to prevent partial writes
    - SHA256 content hashing for change detection
    - Automatic existence checking

    Example:
        resource "pyvider_file_content" "config" {
          filename = "/tmp/app.conf"
          content  = "key=value"
        }

    Attributes:
        filename: Path to the file (relative paths recommended)
        content: Text content to write
        exists: (computed) Whether file exists
        content_hash: (computed) SHA256 hash of content
    """

    async def read(self, ctx: ResourceContext) -> FileContentState | None:
        """
        Read current file state.

        Returns None if the file doesn't exist, triggering Terraform
        to recreate it. This is the correct behavior for resources
        deleted outside of Terraform.

        Args:
            ctx: Resource context with state and config

        Returns:
            Current file state or None if file doesn't exist
        """
        pass

Reference Working Examples

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
"""
File Content Resource

For working examples, see:
https://github.com/provide-io/pyvider-components/tree/main/examples/resource/file_content

Examples include:
- Basic file creation
- Template-based content
- Multi-line configuration files
- Environment-specific files
"""

Common Pitfalls to Avoid

Don't Block on I/O

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# Wrong: Blocking I/O in async function
async def read(self, ctx: ResourceContext) -> State | None:
    data = requests.get(url).json()  # Blocks entire event loop!
    return State(data=data)

# Correct: Use async I/O
async def read(self, ctx: ResourceContext) -> State | None:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        data = response.json()
    return State(data=data)

Don't Store State Outside Terraform

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
# Wrong: External state storage
class MyResource(BaseResource):
    _cache = {}  # Class variable - BAD!

    async def _create(self, ctx: ResourceContext, base_plan: dict) -> tuple[dict | None, None]:
        result = await self.api.create()
        self._cache[result.id] = result  # State leak!
        return {...}, None

# Correct: State only in Terraform
class MyResource(BaseResource):

    async def _create(self, ctx: ResourceContext, base_plan: dict) -> tuple[dict | None, None]:
        result = await self.api.create()
        # Return all state, don't store locally
        return {
            "id": result.id,
            "data": result.data,
        }, None

Don't Make Breaking Schema Changes

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# Wrong: Removing required attribute
@classmethod
def get_schema(cls) -> PvsSchema:
    return s_resource({
        # "filename": a_str(required=True),  # REMOVED - breaks existing configs!
        "path": a_str(required=True),  # NEW NAME - breaking change
    })

# Correct: Add new attribute, deprecate old one
@classmethod
def get_schema(cls) -> PvsSchema:
    return s_resource({
        "filename": a_str(description="(Deprecated) Use 'path' instead"),
        "path": a_str(description="Path to the file"),
        # Support both, migrate users gradually
    })

Don't Ignore Errors

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
# Wrong: Swallowing errors
async def read(self, ctx: ResourceContext) -> State | None:
    try:
        return await self._read_from_api()
    except Exception:
        return None  # User never knows what went wrong

# Correct: Handle specific errors, re-raise unexpected ones
async def read(self, ctx: ResourceContext) -> State | None:
    try:
        return await self._read_from_api()
    except NotFoundError:
        # Expected - resource was deleted
        return None
    except Exception as e:
        # Unexpected - let it propagate with context
        logger.error("Unexpected error reading resource", error=str(e))
        raise

Don't Mix Sync and Async

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# Wrong: Sync in async context
async def _create(self, ctx: ResourceContext, base_plan: dict) -> tuple[dict | None, None]:
    time.sleep(1)  # Blocks!
    result = self.sync_api_call()  # Blocks!
    return {...}, None

# Correct: Async all the way
async def _create(self, ctx: ResourceContext, base_plan: dict) -> tuple[dict | None, None]:
    await asyncio.sleep(1)  # Doesn't block
    result = await self.async_api_call()  # Doesn't block
    return {...}, None

Learn by Example

The best way to learn is by studying working code. Check out pyvider-components for:

  • Production-focused implementations: file_content, local_directory, http_api, and more
  • 100+ working examples: Complete Terraform configurations
  • Comprehensive tests: See how to test every scenario
  • Real-world patterns: Learn from battle-tested code

Remember: The goal is to build providers that are reliable, secure, maintainable, and delightful to use. Follow these best practices, and your users will thank you!