Hook Reference

Complete reference for all DVP CMS plugin hooks. All hooks receive ctx: HookContext as their first parameter (after self) for tenant isolation.

Quick Reference

Hook	Type	Returns	Description
Content Lifecycle
dvp_before_content_create	Filter	dict	Modify content before creation
dvp_content_created	Action	None	React after content created
dvp_before_content_update	Filter	dict	Modify content before update
dvp_content_updated	Action	None	React after content updated
dvp_before_content_delete	Gate	bool	Allow/deny content deletion
dvp_content_deleted	Action	None	React after content deleted
Publishing Lifecycle
dvp_before_content_publish	Gate	PublishGateResult	Quality gate before publish
dvp_content_published	Action	None	React after content published
dvp_before_content_unpublish	Gate	bool	Allow/deny unpublish
dvp_content_unpublished	Action	None	React after unpublished
Commands
dvp_before_command	Filter	Command	Modify command before execution
dvp_after_command	Action	None	React after command executed
dvp_command_failed	First-wins	Any|None	Provide fallback on failure
Events
dvp_before_event_publish	Filter	dict	Modify event before publishing
dvp_after_event_publish	Action	None	React after event published
SEO Metrics
dvp_seo_metrics_received	Filter	SEOMetricsPayload	Enrich incoming SEO data
dvp_before_pattern_extraction	Filter	SEOMetricsPayload	Filter before pattern mining
dvp_pattern_discovered	Action	None	React to pattern discovery
dvp_learning_applied	Action	None	Track learning application
Scheduled Tasks
dvp_register_scheduled_tasks	Collect	list	Register background tasks
dvp_execute_scheduled_task	Execute	TaskResult	Execute a scheduled task
dvp_task_completed	Action	None	React after task completes
dvp_task_failed	First-wins	TaskRetryDecision|None	Control retry behavior
Health Monitoring
dvp_register_health_checks	Collect	list	Register health checks
dvp_execute_health_check	Execute	HealthCheckResult	Execute health check
dvp_health_status_changed	Action	None	React to status changes
dvp_health_anomaly_detected	Action	None	React to anomalies

Which Hook For What?

Use this decision matrix to choose the right hook for your use case:

I want to...	Use this hook	Protocol
Content Modification
Add/modify fields before content is stored	dvp_before_content_create	ContentLifecycleHooks
Transform content on update (add revision, timestamp)	dvp_before_content_update	ContentLifecycleHooks
Add internal links to content	dvp_before_content_create	ContentLifecycleHooks
Validation & Gates
Block low-quality content from publishing	dvp_before_content_publish	ContentLifecycleHooks
Prevent deletion of content with dependencies	dvp_before_content_delete	ContentLifecycleHooks
Require approval before unpublishing	dvp_before_content_unpublish	ContentLifecycleHooks
Notifications & Side Effects
Send Slack/email when content is published	dvp_content_published	ContentLifecycleHooks
Update search index after content changes	dvp_content_created / dvp_content_updated	ContentLifecycleHooks
Invalidate CDN cache on publish	dvp_content_published	ContentLifecycleHooks
Scheduled Work
Run daily cleanup/maintenance tasks	dvp_register_scheduled_tasks	ScheduledTaskHooks
Refresh data from external APIs periodically	dvp_register_scheduled_tasks	ScheduledTaskHooks
Retry failed tasks with custom logic	dvp_task_failed	ScheduledTaskHooks
Monitoring
Monitor domain health/SSL expiry	dvp_register_health_checks	HealthMonitorHooks
Detect anomalies in metrics	dvp_health_anomaly_detected	HealthMonitorHooks
Integrate external SEO data (GSC, Ahrefs)	dvp_seo_metrics_received	SEOMetricsHooks

Hook Types

Filter Hooks

Modify and return a value. The return value replaces the input.

def dvp_before_content_create(
    self,
    ctx: HookContext,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Filter hook - MUST return the modified content."""
    content['processed'] = True
    return content  # Required!

Key Rules:

• Must return a value (never None)
• Returning None raises HookError
• Multiple plugins chain: each receives previous plugin's output

Action Hooks

Perform side effects without returning a value.

def dvp_content_created(
    self,
    ctx: HookContext,
    content_id: str,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> None:
    """Action hook - no return needed."""
    self.logger.info(f"Content created: {content_id}")
    # No return

Key Rules:

• Return value is ignored
• All plugins are called in priority order
• Errors in one plugin don't block others

Gate Hooks

Allow or deny an operation.

def dvp_before_content_delete(
    self,
    ctx: HookContext,
    content_id: str,
) -> bool:
    """Gate hook - return False to block."""
    if self.has_dependencies(content_id):
        return False  # Block deletion
    return True  # Allow deletion

Key Rules:

• Return True to allow, False to block
• If ANY plugin returns False, operation is blocked
• Use for validation, authorization, dependency checks

First-Wins Hooks

First non-None return value is used.

def dvp_command_failed(
    self,
    ctx: HookContext,
    command: Any,
    error: Exception,
) -> Any | None:
    """First-wins hook - first non-None wins."""
    if self.can_recover(error):
        return self.recover(command)
    return None  # Let next plugin try

Key Rules:

• First non-None return stops the chain
• Return None to defer to next plugin
• Use for fallback/recovery logic

Collect Hooks

Aggregate results from all plugins.

def dvp_register_scheduled_tasks(
    self,
    ctx: HookContext,
) -> list[ScheduledTaskDefinition]:
    """Collect hook - all results aggregated."""
    return [
        ScheduledTaskDefinition(
            task_id="my-plugin.cleanup",
            name="Daily Cleanup",
            schedule=TaskSchedule(cron_expression="0 3 * * *"),
        ),
    ]

Key Rules:

• All plugins' results are collected
• Return empty list if no contributions
• Use for registration/discovery patterns

Return Types

These types are used as return values for specific hooks:

PublishGateResult

Returned by dvp_before_content_publish to allow or block publishing:

@dataclass
class PublishGateResult:
    allow: bool                    # True to allow, False to block
    reason: str | None = None      # Human-readable explanation
    blocked_by: str | None = None  # Plugin name that blocked
    details: dict[str, Any] = field(default_factory=dict)  # Additional context

Usage:

def dvp_before_content_publish(self, ctx, content, metadata=None):
    score = content.get('quality_score', 0)
    if score < 70:
        return PublishGateResult(
            allow=False,
            reason=f"Quality score {score} below threshold (70)",
            blocked_by=self.name,
            details={"score": score, "threshold": 70}
        )
    return PublishGateResult(allow=True)

TaskResult

Returned by dvp_execute_scheduled_task:

@dataclass
class TaskResult:
    success: bool                      # Whether task succeeded
    data: dict[str, Any] | None = None # Task output data
    error_message: str | None = None   # Error details if failed
    duration_ms: int | None = None     # Execution time

HealthCheckResult

Returned by dvp_execute_health_check:

@dataclass
class HealthCheckResult:
    status: HealthStatus         # HEALTHY, DEGRADED, UNHEALTHY, UNKNOWN
    message: str | None = None   # Status explanation
    details: dict[str, Any] = field(default_factory=dict)

HookContext

All hooks receive a HookContext object providing tenant and tracing information:

@dataclass
class HookContext:
    tenant_id: UUID           # Tenant identifier
    tenant_slug: str          # Human-readable tenant name
    tenant_plan: str          # Tenant's subscription plan
    correlation_id: UUID | None = None   # Request correlation ID
    causation_id: UUID | None = None     # Event causation ID
    timestamp: datetime | None = None    # Event timestamp
    metadata: dict[str, Any] = field(default_factory=dict)

Usage:

def dvp_content_created(self, ctx, content_id, content, metadata=None):
    # Tenant isolation
    tenant_config = self.get_config(str(ctx.tenant_id))

    # Feature gating by plan
    if ctx.tenant_plan == "enterprise":
        self.enable_advanced_features()

    # Logging with correlation
    self.logger.info(
        f"Content created for {ctx.tenant_slug}",
        extra={"correlation_id": str(ctx.correlation_id)},
    )

Content Lifecycle Hooks

dvp_before_content_create

Called before content is created. Modify content before storage.

def dvp_before_content_create(
    self,
    ctx: HookContext,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """
    Type: Filter hook
    Returns: Modified content dict (never None!)
    """
    content['created_by_plugin'] = self.name
    content['tenant_id'] = str(ctx.tenant_id)
    return content

Priority Recommendations:

• Validation: 1-10 (run first)
• Enhancement: 50-100
• Logging: 900-1000 (run last)

dvp_content_created

Called after content is created. Use for notifications, indexing, etc.

def dvp_content_created(
    self,
    ctx: HookContext,
    content_id: str,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Type: Action hook
    Returns: None
    """
    self.send_notification(f"New content: {content_id}")
    self.update_search_index(ctx.tenant_id, content_id, content)

dvp_before_content_update

Called before content is updated. Modify new content before storage.

def dvp_before_content_update(
    self,
    ctx: HookContext,
    new_content: dict[str, Any],
    content_id: str,
    old_content: dict[str, Any],
) -> dict[str, Any]:
    """
    Type: Filter hook
    Returns: Modified new_content dict
    """
    new_content['updated_at'] = datetime.now().isoformat()
    new_content['revision'] = old_content.get('revision', 0) + 1
    return new_content

dvp_before_content_delete

Called before content deletion. Return False to prevent.

def dvp_before_content_delete(
    self,
    ctx: HookContext,
    content_id: str,
) -> bool:
    """
    Type: Gate hook
    Returns: True to allow, False to block
    """
    if self.has_dependencies(ctx.tenant_id, content_id):
        self.logger.warning(f"Cannot delete {content_id}: has dependencies")
        return False
    return True

Publishing Lifecycle Hooks

dvp_before_content_publish

Quality gate before content is published.

def dvp_before_content_publish(
    self,
    ctx: HookContext,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> PublishGateResult:
    """
    Type: Gate hook
    Returns: PublishGateResult with allow=True/False
    """
    if content.get('quality_score', 0) < 70:
        return PublishGateResult(
            allow=False,
            reason="Quality score below threshold",
            blocked_by=self.name,
        )
    return PublishGateResult(allow=True)

dvp_content_published

Called after content is published. Trigger deployments, invalidate caches.

async def dvp_content_published(
    self,
    ctx: HookContext,
    content_id: str,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Type: Action hook
    Returns: None
    """
    # Trigger deployment
    await self.vercel_client.create_deployment(project_id=self.config.project_id)

    # Invalidate CDN
    domain = content.get('domain')
    self.cdn_client.invalidate(f"/{domain}/*")

Hook Priority

Plugins can specify execution order via get_hook_priority():

class MyPlugin(Plugin, ContentLifecycleHooks):
    def get_hook_priority(self, hook_name: str) -> int:
        """
        Lower number = runs first
        Default: 100
        """
        if hook_name == 'dvp_before_content_create':
            return 10  # Run early (validation)
        return 100  # Default priority

Recommended Priority Ranges:

• Security/Validation: 1-10
• Core processing: 10-50
• Enhancement: 50-100 (default)
• Formatting: 100-200
• Analytics/Metrics: 200-500
• Logging: 900-1000

Implementing Hooks

Basic Pattern

from typing import TYPE_CHECKING, Any
from dvp_cms.plugins.base import Plugin
from dvp_cms.plugins.hookspec import ContentLifecycleHooks

if TYPE_CHECKING:
    from dvp_cms.plugins.context import HookContext


class MyPlugin(Plugin, ContentLifecycleHooks):
    name = "my-plugin"
    version = "1.0.0"

    def dvp_before_content_create(
        self,
        ctx: "HookContext",
        content: dict[str, Any],
        metadata: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        content['processed'] = True
        return content

Async Hooks

Hooks can be sync or async—the system handles both:

async def dvp_content_published(
    self,
    ctx: "HookContext",
    content_id: str,
    content: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> None:
    await self.external_api.notify(content_id)

Testing Hooks

from dvp_cms.plugins.testing import make_test_hook_ctx

def test_hook_modifies_content():
    plugin = MyPlugin()
    ctx = make_test_hook_ctx()

    content = {'title': 'Test'}
    result = plugin.dvp_before_content_create(ctx, content)

    assert result is not None
    assert result['processed'] is True

See Also

• Build Your First Plugin
• Plugin Best Practices
• Testing Plugins
• Scheduled Tasks
• Health Monitoring