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.

Docstrings (Required)

Use Google-style docstrings:

def dvp_before_content_create(
    self,
    content: Dict[str, Any],
    metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    Process content before creation.

    Args:
        content: Content data being created
        metadata: Optional metadata about the operation

    Returns:
        Modified content with processing applied

    Raises:
        ValueError: If content validation fails

    Example:
        >>> plugin = MyPlugin()
        >>> result = plugin.dvp_before_content_create({'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"

    # 2. __init__
    def __init__(self, config: Optional[Dict] = None):
        super().__init__()
        self._validate_config(config)

    # 3. Lifecycle methods
    def initialize(self) -> None:
        super().initialize()

    def shutdown(self) -> None:
        super().shutdown()

    # 4. Public hook implementations
    def dvp_before_content_create(self, content, metadata=None):
        pass

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

    # 6. Private helper methods (prefix with _)
    def _validate_config(self, config: Optional[Dict]) -> None:
        pass

    def _process_text(self, text: str) -> str:
        pass

Security

Input Validation

Always validate external input:

def dvp_before_content_create(self, content, metadata=None):
    # 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

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!

import os

class MyPlugin(Plugin):
    def __init__(self):
        super().__init__()
        self.api_key = os.getenv('MY_PLUGIN_API_KEY')
        if not self.api_key:
            raise ValueError("MY_PLUGIN_API_KEY environment variable required")

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

def dvp_before_content_create(self, content, metadata=None):
    start = time.time()

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

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

    return result

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:

from functools import lru_cache

class MyPlugin(Plugin):
    @lru_cache(maxsize=128)
    def _get_keywords(self, text: str) -> List[str]:
        """Cache keyword extraction results."""
        # Expensive operation here
        return self._extract_keywords(text)

Async Operations

Use async for I/O-bound operations:

import asyncio
import aiohttp

async def _fetch_data(self, url: str) -> Dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

Testing

Test Coverage

Minimum: 80% coverage

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

Test Structure

import pytest
from my_plugin import MyPlugin


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

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

    def test_hook_implementation(self):
        """Test hook modifies content correctly."""
        plugin = MyPlugin()
        content = {'title': 'Test'}
        result = plugin.dvp_before_content_create(content)
        assert 'processed' in result

    def test_error_handling(self):
        """Test plugin handles errors gracefully."""
        plugin = MyPlugin()
        with pytest.raises(ValueError, match="Title required"):
            plugin.dvp_before_content_create({})

    @pytest.mark.asyncio
    async def test_async_operations(self):
        """Test async hook execution."""
        plugin = MyPlugin()
        result = await plugin.async_operation()
        assert result is not None

Error Handling

Graceful Degradation

def dvp_before_content_create(self, content, metadata=None):
    try:
        # Try to enrich content
        enriched_data = self._fetch_enrichment(content)
        content['enriched'] = enriched_data
    except APIError as e:
        # Log error but don't fail
        self.logger.warning(f"Enrichment failed: {e}")
        content['enriched'] = None
    except Exception as e:
        # Unexpected error - log and continue
        self.logger.error(f"Unexpected error in enrichment: {e}", exc_info=True)

    return content

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

Configuration Schema

from typing import Optional, Dict, Any
from dataclasses import dataclass

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

class MyPlugin(Plugin):
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        super().__init__()
        self.config = self._parse_config(config or {})

    def _parse_config(self, config: Dict[str, Any]) -> PluginConfig:
        """Parse and validate configuration."""
        try:
            return PluginConfig(**config)
        except TypeError as e:
            raise ValueError(f"Invalid plugin configuration: {e}")

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/
├── plugin.py           # Main code
├── plugin.json         # Metadata
├── __init__.py         # Package init
├── README.md           # Documentation
├── CHANGELOG.md        # Version history
├── LICENSE             # License file
├── tests/              # Test suite
│   ├── __init__.py
│   └── test_plugin.py
├── pyproject.toml      # Python package config
└── .gitignore          # Git ignore rules

pyproject.toml

[project]
name = "dvp-my-plugin"
version = "1.0.0"
description = "My DVP CMS plugin"
authors = [{name = "Your Name", email = "you@example.com"}]
license = {text = "AGPL-3.0"}
requires-python = ">=3.9"
dependencies = [
    "dvp-cms>=0.1.0",
]

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

[build-system]
requires = ["setuptools>=45", "wheel"]
build-backend = "setuptools.build_meta"

Resources

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