# Reusable Hook Architecture for Django Packages

**Version**: 1.0  
**Last Updated**: 2025-12-22  
**Status**: Production Guide

---

## Overview

This document provides a standardized approach to implementing hook systems across multiple Django packages. It was developed from the `django-wallet-utils` hook system and can be reused for any package requiring extensibility points.

---

## Table of Contents

1. [Core Principles](#core-principles)
2. [Architecture Overview](#architecture-overview)
3. [File Structure](#file-structure)
4. [Implementation Steps](#implementation-steps)
5. [Integration Pattern](#integration-pattern)
6. [Testing Strategy](#testing-strategy)
7. [Documentation Template](#documentation-template)
8. [Package-Specific Adaptation](#package-specific-adaptation)
9. [Example: django-cronjob-utils](#example-django-cronjob-utils)

---

## Core Principles

### 1. **Separation of Concerns**
```
hooks.py          → Core hook system (reusable, generic)
service.py        → Business logic with hook integration
exceptions.py     → Hook-specific exceptions
```

### 2. **Opt-In by Default**
- Hooks should be **disabled by default** for backward compatibility
- Enable explicitly via `enable_hooks=True` parameter

### 3. **Two-Phase Execution**
- **PRE hooks**: Execute before operation (can reject)
- **POST hooks**: Execute after operation (cannot reject, capture errors)

### 4. **Priority-Based Ordering**
- Lower number = higher priority (default: 100)
- Common priorities:
  - `10`: Security/authentication checks
  - `50`: Business rules validation
  - `100`: Default priority
  - `500`: Optional features
  - `900`: Cleanup/logging

### 5. **Immutable Context with Mutable Metadata**
- Transaction parameters are frozen (prevent tampering)
- `metadata` dict allows inter-hook communication

### 6. **Structured Error Codes**
- Use `ALL_CAPS_SNAKE_CASE` format
- Examples: `DAILY_LIMIT_EXCEEDED`, `USER_SUSPENDED`, `INSUFFICIENT_BALANCE`

---

## Architecture Overview

### Component Structure

```
┌─────────────────────────────────────────────────────────┐
│                   Package Service Layer                  │
│  (e.g., WalletService, CronjobService)                  │
└──────────────────────┬──────────────────────────────────┘
                       │
                       │ Calls hooks at lifecycle points
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                    HookManager                           │
│  • execute_pre_hooks() → can reject                     │
│  • execute_post_hooks() → cannot reject                 │
└──────────────────────┬──────────────────────────────────┘
                       │
                       │ Manages execution
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                    HookRegistry                          │
│  • Stores hooks by operation + type                     │
│  • Sorts by priority                                    │
│  • Global singleton instance                            │
└──────────────────────┬──────────────────────────────────┘
                       │
                       │ Retrieves and executes
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│                    Hook Functions                        │
│  • User-defined validation/reaction logic               │
│  • Receives immutable HookContext                       │
│  • Can raise HookRejectionError                         │
└─────────────────────────────────────────────────────────┘
```

---

## File Structure

### Standard Hook System Files

```
your_package/
├── __init__.py               # Export hook functions
├── hooks.py                  # Core hook system (386 lines)
├── exceptions.py             # Hook exceptions (77 lines)
├── service.py                # Service with hook integration
├── models.py                 # Django models
└── repository.py             # Data access layer

examples/
└── hooks/
    ├── README.md             # Hook examples overview
    ├── __init__.py
    ├── example_1.py          # Security/validation hook
    ├── example_2.py          # Business logic hook
    └── example_3.py          # Notification hook

tests/
├── test_hooks.py             # Unit tests for hook system
└── test_hooks_integration.py # Integration tests

docs/
├── usage.md                  # Main documentation with hook section
├── frontend-integration.md   # Error handling for frontend
└── features/
    └── hook-system-design.md # Detailed design document
```

---

## Implementation Steps

### Step 1: Create Core Hook Components

#### File: `hooks.py`

```python
"""Hook system for [package name]."""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from decimal import Decimal
from enum import Enum
from typing import Any, Callable, Dict, List, Optional, Protocol

from .exceptions import HookExecutionError, HookRejectionError

logger = logging.getLogger(__name__)


class HookType(Enum):
    """Hook execution timing types."""
    PRE = "pre"   # Execute before operation
    POST = "post" # Execute after operation


@dataclass(frozen=True)
class HookContext:
    """
    Immutable context passed to hooks.
    
    Customize fields based on your package's operations.
    """
    # Core fields
    operation: str        # e.g., 'add', 'deduct', 'transfer', 'schedule', 'execute'
    
    # Package-specific fields (customize these)
    # Example for wallet:
    # user_id: int
    # point_type: str
    # amount: Decimal
    
    # Example for cronjob:
    # job_id: str
    # schedule: str
    # command: str
    
    # Common fields
    params: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)  # Mutable


@dataclass
class PostHookError:
    """Represents an error in a POST hook."""
    hook_name: str
    error_code: str
    message: str
    details: Dict[str, Any]
    exception: Exception


class HookFunction(Protocol):
    """Protocol for hook functions."""
    def __call__(self, context: HookContext) -> bool | None:
        """
        Execute hook logic.
        
        Returns:
            For PRE: True to continue, False to reject
            For POST: Return value ignored
        """
        ...


@dataclass
class Hook:
    """Represents a registered hook."""
    name: str
    hook_type: HookType
    operation: str  # Operation type or '*' for all
    callback: HookFunction
    priority: int = 100


class HookRegistry:
    """Registry for managing hooks with priority support."""
    
    def __init__(self):
        # Customize operations based on your package
        self._hooks: Dict[str, List[Hook]] = {
            # Example operations - customize these
            "*": [],  # Global hooks
        }
    
    def register(
        self,
        name: str,
        hook_type: HookType,
        callback: HookFunction,
        operation: str = "*",
        priority: int = 100,
    ) -> None:
        """Register a new hook."""
        if operation not in self._hooks:
            raise ValueError(f"Invalid operation: {operation}")
        
        # Check duplicates
        for hooks in self._hooks.values():
            if any(h.name == name for h in hooks):
                raise ValueError(f"Hook '{name}' already registered")
        
        hook = Hook(name, hook_type, operation, callback, priority)
        self._hooks[operation].append(hook)
        
        # Sort by priority (lower = higher priority)
        self._hooks[operation].sort(key=lambda h: (h.priority, h.name))
        
        logger.info(f"Registered {hook_type.value} hook '{name}' for '{operation}' (priority={priority})")
    
    def get_hooks(self, operation: str, hook_type: HookType) -> List[Hook]:
        """Get all hooks for an operation and type, sorted by priority."""
        # Get operation-specific + global hooks
        operation_hooks = [h for h in self._hooks.get(operation, []) if h.hook_type == hook_type]
        global_hooks = [h for h in self._hooks.get("*", []) if h.hook_type == hook_type]
        
        # Combine and sort by priority
        all_hooks = operation_hooks + global_hooks
        all_hooks.sort(key=lambda h: (h.priority, h.name))
        return all_hooks
    
    def unregister(self, name: str) -> bool:
        """Unregister a hook by name."""
        for operation, hooks in self._hooks.items():
            for i, hook in enumerate(hooks):
                if hook.name == name:
                    hooks.pop(i)
                    logger.info(f"Unregistered hook '{name}'")
                    return True
        return False
    
    def clear(self) -> None:
        """Clear all hooks."""
        for operation in self._hooks:
            self._hooks[operation].clear()


class HookManager:
    """Manages hook execution."""
    
    def __init__(self, registry: Optional[HookRegistry] = None):
        self.registry = registry or HookRegistry()
    
    def execute_pre_hooks(self, context: HookContext) -> None:
        """
        Execute PRE hooks. Can reject operation.
        
        Raises:
            HookRejectionError: If any hook rejects
            HookExecutionError: If hook fails unexpectedly
        """
        hooks = self.registry.get_hooks(context.operation, HookType.PRE)
        
        for hook in hooks:
            logger.debug(f"Executing PRE hook '{hook.name}' for {context.operation}")
            
            try:
                result = hook.callback(context)
                
                if result is False:
                    raise HookRejectionError(
                        error_code=f"HOOK_REJECTED_{hook.name.upper()}",
                        message=f"Operation rejected by hook: {hook.name}",
                        details={"hook_name": hook.name},
                    )
                
                logger.debug(f"PRE hook '{hook.name}' completed")
                
            except HookRejectionError:
                raise
            except Exception as e:
                logger.error(f"PRE hook '{hook.name}' failed: {e}", exc_info=True)
                raise HookExecutionError(
                    message=f"Hook '{hook.name}' failed: {str(e)}",
                    hook_name=hook.name,
                    original_exception=e,
                ) from e
    
    def execute_post_hooks(self, context: HookContext) -> List[PostHookError]:
        """
        Execute POST hooks. Errors captured, don't affect success.
        
        Returns:
            List of errors that occurred in POST hooks
        """
        hooks = self.registry.get_hooks(context.operation, HookType.POST)
        errors: List[PostHookError] = []
        
        for hook in hooks:
            logger.debug(f"Executing POST hook '{hook.name}' for {context.operation}")
            
            try:
                hook.callback(context)
                logger.debug(f"POST hook '{hook.name}' completed")
                
            except HookRejectionError as e:
                # POST hooks cannot reject, capture as error
                error = PostHookError(
                    hook_name=hook.name,
                    error_code=e.error_code,
                    message=e.message,
                    details=e.details,
                    exception=e,
                )
                errors.append(error)
                logger.warning(f"POST hook '{hook.name}' raised rejection: {e.message}")
                
            except Exception as e:
                error = PostHookError(
                    hook_name=hook.name,
                    error_code=f"POST_HOOK_ERROR_{hook.name.upper()}",
                    message=str(e),
                    details={},
                    exception=e,
                )
                errors.append(error)
                logger.error(f"POST hook '{hook.name}' failed: {e}", exc_info=True)
        
        return errors


# Global registry singleton
_global_registry = HookRegistry()


def register_hook(
    name: str,
    hook_type: HookType,
    callback: HookFunction,
    operation: str = "*",
    priority: int = 100,
) -> None:
    """Register a hook in the global registry."""
    _global_registry.register(name, hook_type, callback, operation, priority)


def unregister_hook(name: str) -> bool:
    """Unregister a hook from global registry."""
    return _global_registry.unregister(name)


def clear_hooks() -> None:
    """Clear all hooks (useful for testing)."""
    _global_registry.clear()


def get_hook_manager() -> HookManager:
    """Get a hook manager with the global registry."""
    return HookManager(_global_registry)
```

#### File: `exceptions.py`

```python
"""Hook-specific exceptions."""


class HookRejectionError(Exception):
    """
    Raised by PRE hooks to reject an operation with structured error info.
    
    Attributes:
        error_code: Machine-readable code (ALL_CAPS_SNAKE_CASE)
        message: Human-readable error message
        details: Additional context for error handling/UI
    """
    
    def __init__(
        self,
        error_code: str,
        message: str,
        details: dict | None = None,
    ):
        self.error_code = error_code
        self.message = message
        self.details = details or {}
        super().__init__(message)
    
    def __str__(self):
        return f"[{self.error_code}] {self.message}"
    
    def to_dict(self):
        """Convert to dictionary for API responses."""
        return {
            "code": self.error_code,
            "message": self.message,
            "details": self.details,
        }


class HookExecutionError(Exception):
    """
    Raised when a hook fails unexpectedly (not rejection).
    
    Wraps the original exception for debugging.
    """
    
    def __init__(
        self,
        message: str,
        hook_name: str,
        original_exception: Exception,
    ):
        self.message = message
        self.hook_name = hook_name
        self.original_exception = original_exception
        super().__init__(message)
```

---

### Step 2: Define Result Type

Add to your service file:

```python
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from .hooks import PostHookError


@dataclass
class OperationResult:
    """
    Result of an operation with hook execution status.
    
    Customize the success result fields based on your operation.
    """
    success: bool
    
    # Success fields (customize these)
    operation_id: int | None = None  # e.g., transaction_id, job_id, etc.
    
    # Error fields
    error_code: str | None = None
    error_message: str | None = None
    error_details: Dict[str, Any] | None = None
    
    # POST hook errors (operation still successful)
    post_hook_errors: List[PostHookError] | None = None
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for API responses."""
        result = {
            "success": self.success,
            "operation_id": self.operation_id,
        }
        
        if self.error_code:
            result["error"] = {
                "code": self.error_code,
                "message": self.error_message,
                "details": self.error_details or {},
            }
        
        if self.post_hook_errors:
            result["warnings"] = [
                {
                    "hook": err.hook_name,
                    "code": err.error_code,
                    "message": err.message,
                    "details": err.details,
                }
                for err in self.post_hook_errors
            ]
        
        return result
```

---

### Step 3: Integrate Hooks into Service

```python
class YourService:
    """Service with hook integration."""
    
    def __init__(
        self,
        repository: YourRepository,
        enable_hooks: bool = False,  # Default disabled for backward compatibility
        hook_manager: Optional[HookManager] = None,
    ):
        self.repository = repository
        self.enable_hooks = enable_hooks
        self.hook_manager = hook_manager or get_hook_manager()
    
    def perform_operation(
        self,
        # Your operation parameters
        param1: str,
        param2: int,
    ) -> int | OperationResult:
        """
        Perform operation with optional hook support.
        
        Returns:
            If hooks disabled: operation_id (int)
            If hooks enabled: OperationResult
        """
        if self.enable_hooks:
            return self._perform_operation_with_hooks(param1, param2)
        
        # Original implementation (backward compatible)
        result_id = self._do_operation(param1, param2)
        return result_id
    
    def _perform_operation_with_hooks(
        self,
        param1: str,
        param2: int,
    ) -> OperationResult:
        """Perform operation with hooks enabled."""
        
        # 1. Create hook context
        context = HookContext(
            operation="your_operation",
            # Add your operation-specific fields
            # param1=param1,
            # param2=param2,
            params={"param1": param1, "param2": param2},
        )
        
        # 2. Execute PRE hooks
        try:
            self.hook_manager.execute_pre_hooks(context)
        except HookRejectionError as e:
            return OperationResult(
                success=False,
                error_code=e.error_code,
                error_message=e.message,
                error_details=e.details,
            )
        except HookExecutionError as e:
            return OperationResult(
                success=False,
                error_code="HOOK_EXECUTION_ERROR",
                error_message=str(e),
                error_details={"hook_name": e.hook_name},
            )
        
        # 3. Perform actual operation
        try:
            operation_id = self._do_operation(param1, param2)
        except Exception as e:
            # Handle operation-specific errors
            return OperationResult(
                success=False,
                error_code="OPERATION_FAILED",
                error_message=str(e),
            )
        
        # 4. Execute POST hooks
        post_errors = self.hook_manager.execute_post_hooks(context)
        
        # 5. Return result
        return OperationResult(
            success=True,
            operation_id=operation_id,
            post_hook_errors=post_errors if post_errors else None,
        )
    
    def _do_operation(self, param1: str, param2: int) -> int:
        """Actual operation implementation."""
        # Your business logic here
        pass
```

---

## Integration Pattern

### Pattern 1: Service-Level Integration (Recommended)

**Pros**: Centralized, easy to test, consistent behavior  
**Cons**: Requires service layer

```python
# In your service __init__
def __init__(self, enable_hooks: bool = False):
    self.enable_hooks = enable_hooks
    self.hook_manager = get_hook_manager() if enable_hooks else None

# In each operation method
if self.enable_hooks:
    return self._operation_with_hooks(...)
return self._operation_without_hooks(...)
```

### Pattern 2: Decorator Pattern

**Pros**: Clean, reusable, minimal boilerplate  
**Cons**: Less explicit, harder to test

```python
def with_hooks(operation_name: str):
    """Decorator to add hook support to methods."""
    def decorator(func):
        def wrapper(self, *args, **kwargs):
            if not self.enable_hooks:
                return func(self, *args, **kwargs)
            
            # Create context, execute PRE hooks
            # Execute function
            # Execute POST hooks
            # Return result
        return wrapper
    return decorator

# Usage
@with_hooks("schedule_job")
def schedule_job(self, ...):
    pass
```

### Pattern 3: Mixin Pattern

**Pros**: Separates hook logic, reusable across services  
**Cons**: More complex inheritance

```python
class HookMixin:
    """Mixin providing hook execution capabilities."""
    
    def execute_with_hooks(self, operation: str, func, *args, **kwargs):
        # Hook execution logic
        pass

class YourService(HookMixin):
    def operation(self, ...):
        if self.enable_hooks:
            return self.execute_with_hooks("operation", self._do_operation, ...)
        return self._do_operation(...)
```

---

## Testing Strategy

### Unit Tests for Hook System

```python
# tests/test_hooks.py

import pytest
from your_package.hooks import (
    HookRegistry, HookManager, HookType, HookContext,
    register_hook, unregister_hook, clear_hooks
)
from your_package.exceptions import HookRejectionError, HookExecutionError


@pytest.fixture
def clean_registry():
    """Clear hooks before each test."""
    clear_hooks()
    yield
    clear_hooks()


def test_register_and_execute_pre_hook(clean_registry):
    """Test registering and executing a PRE hook."""
    executed = []
    
    def test_hook(context: HookContext) -> bool:
        executed.append(context.operation)
        return True
    
    register_hook("test", HookType.PRE, test_hook, "test_operation")
    
    manager = HookManager()
    context = HookContext(operation="test_operation")
    
    manager.execute_pre_hooks(context)
    
    assert executed == ["test_operation"]


def test_pre_hook_rejection(clean_registry):
    """Test PRE hook can reject operation."""
    def reject_hook(context: HookContext) -> bool:
        raise HookRejectionError(
            error_code="TEST_REJECTION",
            message="Test rejection",
        )
    
    register_hook("reject", HookType.PRE, reject_hook)
    
    manager = HookManager()
    context = HookContext(operation="test")
    
    with pytest.raises(HookRejectionError) as exc:
        manager.execute_pre_hooks(context)
    
    assert exc.value.error_code == "TEST_REJECTION"


def test_hook_priority_ordering(clean_registry):
    """Test hooks execute in priority order."""
    execution_order = []
    
    def make_hook(name: str):
        def hook(ctx: HookContext):
            execution_order.append(name)
            return True
        return hook
    
    register_hook("low", HookType.PRE, make_hook("low"), priority=500)
    register_hook("high", HookType.PRE, make_hook("high"), priority=10)
    register_hook("mid", HookType.PRE, make_hook("mid"), priority=100)
    
    manager = HookManager()
    manager.execute_pre_hooks(HookContext(operation="test"))
    
    assert execution_order == ["high", "mid", "low"]


def test_post_hooks_capture_errors(clean_registry):
    """Test POST hooks capture errors without failing."""
    def error_hook(context: HookContext):
        raise ValueError("Test error")
    
    register_hook("error", HookType.POST, error_hook)
    
    manager = HookManager()
    errors = manager.execute_post_hooks(HookContext(operation="test"))
    
    assert len(errors) == 1
    assert errors[0].hook_name == "error"
    assert "Test error" in errors[0].message
```

### Integration Tests

```python
# tests/test_hooks_integration.py

def test_service_with_hooks_enabled():
    """Test service operation with hooks enabled."""
    service = YourService(enable_hooks=True)
    
    # Register validation hook
    def validate(context: HookContext) -> bool:
        if context.params.get("invalid"):
            raise HookRejectionError("INVALID_PARAM", "Invalid parameter")
        return True
    
    register_hook("validate", HookType.PRE, validate)
    
    # Test successful operation
    result = service.perform_operation(param1="valid", param2=123)
    assert result.success is True
    assert result.operation_id is not None
    
    # Test rejected operation
    result = service.perform_operation(param1="invalid", param2=123, invalid=True)
    assert result.success is False
    assert result.error_code == "INVALID_PARAM"
```

---

## Documentation Template

### For `docs/usage.md`

```markdown
## Hook System

The hook system allows you to register custom logic that executes before and after operations.

### Quick Example

\`\`\`python
from your_package import YourService, register_hook, HookType, HookContext
from your_package.exceptions import HookRejectionError

# Define a validation hook
def check_limit(context: HookContext) -> bool:
    if context.params["amount"] > 1000:
        raise HookRejectionError(
            error_code="LIMIT_EXCEEDED",
            message="Amount exceeds limit",
            details={"limit": 1000, "requested": context.params["amount"]}
        )
    return True

# Register the hook
register_hook(
    name="limit_check",
    hook_type=HookType.PRE,
    callback=check_limit,
    operation="your_operation",
    priority=50
)

# Use service with hooks enabled
service = YourService(enable_hooks=True)
result = service.perform_operation(amount=1500)

if not result.success:
    print(f"Error: {result.error_code} - {result.error_message}")
\`\`\`

### Hook Types

- **PRE hooks**: Execute before operation, can reject
- **POST hooks**: Execute after operation, cannot reject

### Priority System

Lower number = higher priority (default: 100)

- `10`: Security/authentication
- `50`: Business validation
- `100`: Default
- `500`: Optional features
- `900`: Logging/cleanup

### Error Handling

PRE hooks can reject operations:

\`\`\`python
raise HookRejectionError(
    error_code="ALL_CAPS_ERROR_CODE",
    message="Human-readable message",
    details={"key": "value"}
)
\`\`\`

POST hook errors are captured but don't fail the operation:

\`\`\`python
result = service.operation(...)
if result.post_hook_errors:
    for error in result.post_hook_errors:
        print(f"Warning: {error.hook_name} - {error.message}")
\`\`\`

### Best Practices

1. ✅ Use PRE hooks for validation
2. ✅ Use POST hooks for notifications/logging
3. ✅ Return structured error codes
4. ✅ Keep hooks fast (<100ms recommended)
5. ❌ Don't modify transaction parameters
6. ❌ Don't depend on external state in PRE hooks
```

---

## Package-Specific Adaptation

### Key Customization Points

1. **HookContext fields**: Define based on your operations
2. **Operations list**: Define valid operation types
3. **Result type fields**: Customize success result data
4. **Error codes**: Define package-specific codes
5. **Examples**: Create 3-5 example hooks for your domain

### Checklist for New Package

- [ ] Copy `hooks.py` and customize `HookContext`
- [ ] Copy `exceptions.py` (usually no changes needed)
- [ ] Define operations in `HookRegistry.__init__()`
- [ ] Create `OperationResult` dataclass in service
- [ ] Integrate hooks into service methods
- [ ] Add `enable_hooks` parameter (default=False)
- [ ] Write 10+ unit tests for hook system
- [ ] Write 5+ integration tests
- [ ] Create 3-5 example hooks
- [ ] Document in `usage.md`
- [ ] Document error codes for frontend

---

## Example: django-cronjob-utils

Here's how to adapt the hook system for a cronjob management package:

### 1. Define HookContext

```python
@dataclass(frozen=True)
class CronjobHookContext:
    """Context for cronjob hooks."""
    
    # Operation type
    operation: str  # 'schedule', 'execute', 'pause', 'resume', 'delete'
    
    # Job details
    job_id: str
    schedule: str  # cron expression
    command: str
    
    # Execution context (for 'execute' operation)
    execution_id: Optional[str] = None
    last_run: Optional[str] = None
    next_run: Optional[str] = None
    
    # Additional context
    params: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)
```

### 2. Define Operations

```python
class HookRegistry:
    def __init__(self):
        self._hooks: Dict[str, List[Hook]] = {
            "schedule": [],  # Creating a new job
            "execute": [],   # Job execution
            "pause": [],     # Pausing a job
            "resume": [],    # Resuming a job
            "delete": [],    # Deleting a job
            "*": [],         # All operations
        }
```

### 3. Define Result Type

```python
@dataclass
class CronjobResult:
    """Result of a cronjob operation."""
    success: bool
    job_id: str | None = None
    execution_id: str | None = None  # For execute operations
    error_code: str | None = None
    error_message: str | None = None
    error_details: Dict[str, Any] | None = None
    post_hook_errors: List[PostHookError] | None = None
```

### 4. Example Hooks

```python
# examples/hooks/max_jobs_limit.py
"""Limit maximum number of scheduled jobs per user."""

from cronjob_utils import CronjobHookContext, HookType, register_hook
from cronjob_utils.exceptions import HookRejectionError

def check_job_limit(context: CronjobHookContext) -> bool:
    """Prevent users from scheduling too many jobs."""
    if context.operation != "schedule":
        return True
    
    user_id = context.params.get("user_id")
    max_jobs = context.params.get("max_jobs_per_user", 10)
    
    # Get current job count
    current_count = get_user_job_count(user_id)
    
    if current_count >= max_jobs:
        raise HookRejectionError(
            error_code="MAX_JOBS_EXCEEDED",
            message=f"Maximum {max_jobs} jobs per user exceeded",
            details={
                "current_count": current_count,
                "max_allowed": max_jobs,
            }
        )
    
    return True

# Register with high priority (security check)
register_hook(
    name="max_jobs_limit",
    hook_type=HookType.PRE,
    callback=check_job_limit,
    operation="schedule",
    priority=10,
)
```

```python
# examples/hooks/execution_timeout_check.py
"""Check if job execution is taking too long."""

from datetime import datetime, timedelta
from cronjob_utils import CronjobHookContext, HookType, register_hook
from cronjob_utils.exceptions import HookRejectionError

def check_execution_timeout(context: CronjobHookContext) -> bool:
    """Prevent job from running if previous execution still running."""
    if context.operation != "execute":
        return True
    
    job_id = context.job_id
    timeout_minutes = context.params.get("timeout_minutes", 60)
    
    # Check if previous execution still running
    last_execution = get_last_execution(job_id)
    
    if last_execution and not last_execution.completed:
        started_at = last_execution.started_at
        now = datetime.now()
        
        if (now - started_at) > timedelta(minutes=timeout_minutes):
            # Mark as timeout and allow new execution
            mark_execution_timeout(last_execution.id)
            context.metadata["previous_timeout"] = True
        else:
            # Still running, reject
            raise HookRejectionError(
                error_code="EXECUTION_IN_PROGRESS",
                message="Previous execution still running",
                details={
                    "execution_id": last_execution.id,
                    "started_at": started_at.isoformat(),
                    "timeout_in_seconds": (timeout_minutes * 60) - (now - started_at).seconds,
                }
            )
    
    return True

register_hook(
    name="execution_timeout_check",
    hook_type=HookType.PRE,
    callback=check_execution_timeout,
    operation="execute",
    priority=50,
)
```

```python
# examples/hooks/job_execution_logger.py
"""Log job executions for audit trail."""

from cronjob_utils import CronjobHookContext, HookType, register_hook
import logging

logger = logging.getLogger(__name__)

def log_execution(context: CronjobHookContext) -> None:
    """Log job execution to audit trail."""
    logger.info(
        f"Job executed: {context.job_id}",
        extra={
            "job_id": context.job_id,
            "execution_id": context.execution_id,
            "command": context.command,
            "schedule": context.schedule,
            "metadata": context.metadata,
        }
    )

register_hook(
    name="execution_logger",
    hook_type=HookType.POST,
    callback=log_execution,
    operation="execute",
    priority=900,  # Low priority (logging)
)
```

### 5. Service Integration

```python
class CronjobService:
    def __init__(
        self,
        repository: CronjobRepository,
        enable_hooks: bool = False,
        hook_manager: Optional[HookManager] = None,
    ):
        self.repository = repository
        self.enable_hooks = enable_hooks
        self.hook_manager = hook_manager or get_hook_manager()
    
    def schedule_job(
        self,
        job_id: str,
        schedule: str,
        command: str,
        user_id: int,
    ) -> str | CronjobResult:
        """Schedule a new cronjob."""
        if self.enable_hooks:
            return self._schedule_job_with_hooks(job_id, schedule, command, user_id)
        
        # Original implementation
        return self._do_schedule(job_id, schedule, command)
    
    def _schedule_job_with_hooks(
        self,
        job_id: str,
        schedule: str,
        command: str,
        user_id: int,
    ) -> CronjobResult:
        """Schedule job with hooks."""
        
        # Create context
        context = CronjobHookContext(
            operation="schedule",
            job_id=job_id,
            schedule=schedule,
            command=command,
            params={"user_id": user_id},
        )
        
        # PRE hooks
        try:
            self.hook_manager.execute_pre_hooks(context)
        except HookRejectionError as e:
            return CronjobResult(
                success=False,
                error_code=e.error_code,
                error_message=e.message,
                error_details=e.details,
            )
        
        # Execute
        job_id = self._do_schedule(job_id, schedule, command)
        
        # POST hooks
        post_errors = self.hook_manager.execute_post_hooks(context)
        
        return CronjobResult(
            success=True,
            job_id=job_id,
            post_hook_errors=post_errors if post_errors else None,
        )
```

---

## Summary Checklist

When implementing hooks in a new package:

### Planning Phase
- [ ] Identify operations that need hooks
- [ ] Define HookContext fields
- [ ] Define result type
- [ ] List common error codes
- [ ] Plan 3-5 example use cases

### Implementation Phase
- [ ] Copy and customize `hooks.py` (~400 lines)
- [ ] Copy `exceptions.py` (~80 lines)
- [ ] Create result dataclass (~50 lines)
- [ ] Integrate into service (~100 lines per operation)
- [ ] Set `enable_hooks=False` by default

### Testing Phase
- [ ] Unit tests for hook registry (5+ tests)
- [ ] Unit tests for hook manager (5+ tests)
- [ ] Integration tests (5+ scenarios)
- [ ] Test priority ordering
- [ ] Test error handling (PRE and POST)

### Documentation Phase
- [ ] Add Hook System section to usage.md
- [ ] Document all error codes
- [ ] Create 3-5 example hooks
- [ ] Document priority guidelines
- [ ] Create frontend integration guide (if applicable)

### Quality Assurance
- [ ] All existing tests still pass
- [ ] Backward compatibility maintained
- [ ] Performance impact < 5ms per operation
- [ ] Hook examples are production-ready
- [ ] Documentation is clear and complete

---

## References

**Source Package**: `django-wallet-utils`  
**Lines of Code**:
- `hooks.py`: 386 lines
- `exceptions.py`: 77 lines
- `service.py` integration: ~200 lines
- Tests: 1,181 lines (44 tests)
- Examples: 1,576 lines (5 examples)
- Documentation: ~3,000 lines

**Key Files to Reference**:
- `/src/wallet_utils/hooks.py` - Core hook system
- `/src/wallet_utils/service.py` - Service integration pattern
- `/src/wallet_utils/exceptions.py` - Exception definitions
- `/tests/test_hooks.py` - Unit test patterns
- `/tests/test_hooks_integration.py` - Integration test patterns
- `/examples/hooks/` - Production-ready examples
- `/docs/usage.md` - User documentation

---

**Document maintained by**: Development Team  
**Next Review**: When implementing third package with hooks