Plugin Development Best Practices

This guide covers best practices for building production-ready DVP CMS plugins.

Code Quality

Type Hints (Required)

def process_content(content):
    return content

def process_content(content: dict[str, Any]) -> dict[str, Any]:
    return content

Why: Type hints enable IDE autocomplete, catch errors early, and serve as documentation. Use Python 3.12+ syntax (dict instead of Dict).

Docstrings (Required)

Use Google-style docstrings:

def dvp_before_content_create(
    self,
    ctx: HookContext,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Process content before creation.

    This is a FILTER hook - ctx is always first, must return a value.
    All hooks receive ctx as first parameter for consistency.

    Args:
        ctx: Hook context with tenant information (always first)
        content: Content data being created
        metadata: Optional metadata for the operation

    Returns:
        Modified content dict. Filter hooks must return a value!

    Raises:
        ValueError: If content validation fails

    Example:
        >>> plugin = MyPlugin()
        >>> ctx = create_test_context()
        >>> result = plugin.dvp_before_content_create(ctx, {'title': 'Test'})
        >>> print(result['processed'])
        True
    """

Code Organization

class MyPlugin(Plugin, ContentLifecycleHooks):
    """Plugin for..."""

    # 1. Class attributes (metadata)
    name = "my-plugin"
    version = "1.0.0"
    description = "Does something useful"
    author = "Your Name"

    # 2. __init__ (simple, no parameters)
    def __init__(self) -> None:
        super().__init__()
        self._cache: dict[str, Any] = {}

    # 3. Hook implementations
    # ALL hooks: ctx ALWAYS first for consistency
    # FILTER hooks: must return value
    def dvp_before_content_create(
        self,
        ctx: HookContext,
        content: dict[str, Any],
        metadata: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        return content  # Must return!

    # ACTION hooks: ctx first, no return value
    def dvp_content_published(
        self,
        ctx: HookContext,
        content_id: str,
        content: dict[str, Any],
        metadata: dict[str, Any] | None = None,
    ) -> None:
        pass  # Side effects only

    # 4. Public utility methods
    def get_statistics(self) -> dict[str, Any]:
        pass

    # 5. Private helper methods (prefix with _)
    def _validate_input(self, data: dict[str, Any]) -> None:
        pass

    async def _fetch_external_data(self, url: str) -> dict[str, Any]:
        pass

Security

Input Validation

Always validate external input:

def dvp_before_content_create(
    self,
    ctx: HookContext,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Validate and sanitize content before creation.

    All hooks receive ctx as first parameter for consistency.
    """
    # Validate required fields
    if 'title' not in content:
        raise ValueError("Title is required")

    # Validate types
    if not isinstance(content['title'], str):
        raise TypeError("Title must be a string")

    # Sanitize input
    content['title'] = self._sanitize_html(content['title'])

    return content  # Filter hooks MUST return a value

SQL Injection Prevention

# DANGEROUS!
query = f"SELECT * FROM content WHERE title = '{content['title']}'"

# Safe
query = "SELECT * FROM content WHERE title = %s"
cursor.execute(query, (content['title'],))

Path Traversal Prevention

file_path = f"/uploads/{content['filename']}"  # Can be ../../etc/passwd

from pathlib import Path

def get_safe_path(self, filename: str) -> Path:
    # Remove path components
    safe_name = Path(filename).name
    base_dir = Path("/uploads")
    full_path = (base_dir / safe_name).resolve()

    # Ensure path is within base_dir
    if not str(full_path).startswith(str(base_dir)):
        raise ValueError("Invalid filename")

    return full_path

Secret Management

API_KEY = "sk-1234567890abcdef"  # NO!

def dvp_before_content_create(
    self,
    ctx: HookContext,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    # Get tenant-specific secrets (encrypted storage)
    secrets = self.get_secrets(str(ctx.tenant_id))
    api_key = secrets.require("api_key")  # Raises if missing
    optional_token = secrets.get("optional_token")  # None if missing
    # ... modify content using secrets ...
    return content

XSS Prevention

import html

def _sanitize_html(self, text: str) -> str:
    """Escape HTML to prevent XSS."""
    return html.escape(text)

Performance

Hook Execution Time

Target: <100ms per hook execution

import time
from dvp_cms.kernel.logging import get_logger

logger = get_logger(__name__)

def dvp_before_content_create(
    self,
    ctx: HookContext,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Filter hook with performance monitoring.

    All hooks receive ctx as first parameter for consistency.
    """
    start = time.time()

    # Your processing here
    result = self._process(content)

    elapsed = time.time() - start
    if elapsed > 0.1:  # 100ms
        logger.warning(f"Hook execution slow: {elapsed:.3f}s")

    return result  # Filter hooks MUST return a value

Database Queries

for content_id in content_ids:
    content = db.query(Content).filter_by(id=content_id).first()  # Bad!

contents = db.query(Content).filter(Content.id.in_(content_ids)).all()

Caching

Cache expensive operations per tenant:

class MyPlugin(Plugin, ContentLifecycleHooks):
    def __init__(self) -> None:
        super().__init__()
        self._cache: dict[str, tuple[datetime, Any]] = {}
        self._cache_ttl = 3600  # 1 hour

    def _get_cached(self, key: str) -> Any | None:
        """Get cached value if not expired."""
        if key in self._cache:
            cached_at, value = self._cache[key]
            if (datetime.now() - cached_at).total_seconds() < self._cache_ttl:
                return value
        return None

    def _set_cached(self, key: str, value: Any) -> None:
        """Cache a value with timestamp."""
        self._cache[key] = (datetime.now(), value)

Async Operations

Use async with httpx for I/O-bound operations:

import httpx

async def _fetch_data(self, url: str, headers: dict[str, str]) -> dict[str, Any]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.get(url, headers=headers)
        response.raise_for_status()
        return response.json()

Testing

Test Coverage

Minimum: 80% coverage

pytest --cov=my_plugin --cov-report=term-missing

Test Structure

from uuid import UUID
import pytest
from dvp_cms.plugins.context import HookContext
from my_plugin import MyPlugin

TENANT_ID = UUID("11111111-1111-1111-1111-111111111111")


@pytest.fixture
def hook_context() -> HookContext:
    """Create a minimal HookContext for testing."""
    return HookContext(
        tenant_id=TENANT_ID,
        tenant_slug="test-tenant",
        tenant_plan="enterprise",
    )


class TestMyPlugin:
    """Test suite for MyPlugin."""

    def test_plugin_metadata(self) -> None:
        """Test plugin has correct metadata."""
        plugin = MyPlugin()
        assert plugin.name == "my-plugin"
        assert plugin.version == "1.0.0"

    def test_filter_hook_implementation(
        self, hook_context: HookContext
    ) -> None:
        """Test filter hook modifies content correctly.

        All hooks receive ctx as first parameter for consistency.
        """
        plugin = MyPlugin()
        content = {'title': 'Test'}
        # All hooks: ctx first
        result = plugin.dvp_before_content_create(hook_context, content)
        assert result is not None  # Filter hooks must return a value
        assert 'processed' in result

    def test_filter_hook_error_handling(self, hook_context: HookContext) -> None:
        """Test filter hook handles errors gracefully."""
        plugin = MyPlugin()
        # All hooks: ctx first
        with pytest.raises(ValueError, match="Title required"):
            plugin.dvp_before_content_create(hook_context, {})

Error Handling

Graceful Degradation

from dvp_cms.kernel.logging import get_logger

logger = get_logger(__name__)

def dvp_before_content_create(
    self,
    ctx: HookContext,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Filter hook with graceful error handling.

    All hooks receive ctx first for consistency. Filter hooks must return a value.
    Even on error, we return the content (possibly without enrichment).
    """
    try:
        # Try to enrich content
        enriched_data = self._fetch_enrichment(content)
        content['enriched'] = enriched_data
    except httpx.HTTPStatusError as e:
        # Log error but don't fail
        logger.warning(f"Enrichment failed: {e}")
        content['enriched'] = None
    except Exception as e:
        # Unexpected error - log and continue
        logger.exception(f"Unexpected error: {e}")

    return content  # Filter hooks MUST return a value

Helpful Error Messages

raise ValueError("Invalid input")

raise ValueError(
    f"Invalid content: missing required field 'title'. "
    f"Received fields: {list(content.keys())}"
)

Hook Priorities

Priority Ranges

Use semantic priority ranges:

def get_hook_priority(self, hook_name: str) -> int:
    """
    Priority ranges:
    1-10:     Security, authentication (highest)
    10-50:    Validation, sanitization
    50-100:   Content transformation
    100-500:  Enhancement, enrichment
    500-900:  Formatting, presentation
    900-1000: Logging, analytics (lowest)
    """
    if hook_name in ['dvp_before_content_create', 'dvp_before_content_update']:
        return 60  # Transformation
    elif hook_name == 'dvp_content_created':
        return 950  # Logging
    return 500  # Default

Configuration

Tenant-Scoped Configuration

from dataclasses import dataclass
from typing import Any

@dataclass(frozen=True)
class PluginConfig:
    """Configuration for MyPlugin."""
    api_endpoint: str
    timeout: int = 30
    retry_count: int = 3
    cache_enabled: bool = True

    def __post_init__(self) -> None:
        if self.timeout < 1:
            raise ValueError("timeout must be positive")

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "PluginConfig":
        return cls(
            api_endpoint=data["api_endpoint"],
            timeout=data.get("timeout", 30),
            retry_count=data.get("retry_count", 3),
            cache_enabled=data.get("cache_enabled", True),
        )


class MyPlugin(Plugin, ContentLifecycleHooks):
    def __init__(self) -> None:
        super().__init__()
        self._configs: dict[str, PluginConfig] = {}

    def _get_config(self, tenant_id: str) -> PluginConfig:
        """Get or create cached config for tenant."""
        if tenant_id not in self._configs:
            config_dict = self.get_config(tenant_id)
            self._configs[tenant_id] = PluginConfig.from_dict(config_dict)
        return self._configs[tenant_id]

Versioning

Semantic Versioning

Follow semver:

• Major (1.0.0): Breaking changes
• Minor (0.1.0): New features, backward compatible
• Patch (0.0.1): Bug fixes

class MyPlugin(Plugin):
    name = "my-plugin"
    version = "1.2.3"  # MAJOR.MINOR.PATCH

Deployment

Package Structure

my-plugin/
├── pyproject.toml      # Package metadata and entry point
├── README.md           # Documentation
├── CHANGELOG.md        # Version history
├── LICENSE             # License file
├── src/
│   └── my_plugin/
│       ├── __init__.py     # Package exports
│       ├── plugin.py       # Main plugin code
│       ├── config.py       # Configuration dataclass
│       └── client.py       # API client (if needed)
└── tests/
    └── test_plugin.py

pyproject.toml

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "dvp-my-plugin"
version = "1.0.0"
description = "My DVP CMS plugin"
readme = "README.md"
license = "MIT"
requires-python = ">=3.12"
authors = [
    { name = "Your Name", email = "you@example.com" },
]
dependencies = [
    "dvp-cms>=1.0.0",
]

[project.entry-points."dvp_cms.plugins"]
my-plugin = "my_plugin:MyPlugin"

[tool.hatch.build.targets.wheel]
packages = ["src/my_plugin"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

Resources

• Style Guide: PEP 8
• Type Hints: PEP 484
• Docstrings: Google Style Guide
• Testing: pytest Documentation
• Versioning: Semantic Versioning