Async vs Sync Hooks

When to use async hooks and how to implement them correctly.

Overview

DVP CMS hooks can be either synchronous or asynchronous. The choice depends on:

• Whether the operation performs I/O (network, disk)
• Whether blocking would impact performance
• The hook's intended use case

Rule of Thumb:

• Sync hooks: CPU-bound operations, quick transformations, validation
• Async hooks: Network calls, database queries, file I/O, long-running operations

Decision Matrix

Scenario	Use Sync	Use Async
Validate input data	✓
Transform content	✓
Calculate values	✓
Check permissions (local)	✓
HTTP API calls		✓
Database queries		✓
File I/O		✓
Send notifications		✓
Deploy to CDN		✓

Sync Hook Implementation

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 ValidationPlugin(Plugin, ContentLifecycleHooks):
    """Plugin with sync hooks for validation."""

    name = "validation"
    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]:
        """Sync hook - quick validation and transformation."""
        # CPU-bound operations are fine
        if not content.get("title"):
            content["title"] = "Untitled"

        # Add computed fields
        content["word_count"] = len(content.get("body", "").split())

        return content

Async Hook Implementation

class NotificationPlugin(Plugin, ContentLifecycleHooks):
    """Plugin with async hooks for I/O operations."""

    name = "notification"
    version = "1.0.0"

    async def dvp_content_published(
        self,
        ctx: "HookContext",
        content_id: str,
        content: dict[str, Any],
        metadata: dict[str, Any] | None = None,
    ) -> None:
        """Async hook - sends notifications over network."""
        import httpx

        # Network I/O - must be async
        async with httpx.AsyncClient() as client:
            await client.post(
                self._webhook_url,
                json={"text": f"Content published: {content.get('title')}"},
            )

Common Mistakes

Blocking in Async Context

# WRONG - blocking call in async hook
async def dvp_content_published(self, ctx, content_id, content, metadata=None):
    import requests  # Blocking library!
    requests.post(url, data=content)  # Blocks event loop!

# RIGHT - use async HTTP client
async def dvp_content_published(self, ctx, content_id, content, metadata=None):
    import httpx
    async with httpx.AsyncClient() as client:
        await client.post(url, json=content)

Forgetting to Await

# WRONG - missing await
async def dvp_content_published(self, ctx, content_id, content, metadata=None):
    self._send_notification(content_id)  # Returns coroutine, doesn't execute!

# RIGHT - await the coroutine
async def dvp_content_published(self, ctx, content_id, content, metadata=None):
    await self._send_notification(content_id)

Performance Patterns

Parallel Async Operations

async def dvp_content_published(self, ctx, content_id, content, metadata=None):
    import asyncio

    # Run independent operations in parallel
    await asyncio.gather(
        self._notify_slack(content_id),
        self._update_search_index(content_id),
        self._log_analytics(content_id),
    )

Timeout Protection

async def dvp_execute_health_check(self, check, ctx, execution_ctx):
    import asyncio

    try:
        result = await asyncio.wait_for(
            self._check_domain(execution_ctx.domain),
            timeout=check.timeout_seconds,
        )
        return HealthCheckResult(
            domain=execution_ctx.domain,
            overall_status=HealthStatus.HEALTHY,
        )
    except asyncio.TimeoutError:
        return HealthCheckResult(
            domain=execution_ctx.domain,
            overall_status=HealthStatus.UNHEALTHY,
            details=[HealthCheckDetail(message="Check timed out")],
        )

See Also

• Hook Reference
• Plugin Best Practices
• Troubleshooting