Hook System Reference

on_<category>_<timing>_<operation>

Examples:
- on_record_before_create
- on_auth_after_login
- on_collection_before_delete

from snackbase.core.hooks import HookEvent

@app.state.hook.on_record_before_create("posts", priority=100)
async def validate_post(event, data, context):
    """Validate post before creation."""
    if data and len(data.get("title", "")) < 5:
        from snackbase.domain.entities.hook_context import AbortHookException
        raise AbortHookException("Title must be at least 5 characters")
    return data

@dataclass
class HookContext:
    """Context passed to hooks."""
    app: Any                      # FastAPI app instance
    user: Optional["User"]        # Current authenticated user
    account_id: Optional[str]     # Current account ID
    request_id: str               # Request correlation ID
    request: Optional["Request"]  # FastAPI/Starlette Request object
    ip_address: Optional[str]     # Client IP address
    user_agent: Optional[str]     # Client user agent
    user_name: Optional[str]      # User display name

from snackbase.domain.entities.hook_context import AbortHookException

@app.state.hook.on_record_before_create("posts")
async def validate_post(event, data, context):
    """Prevent spam posts."""
    if data and "spam" in data.get("content", "").lower():
        raise AbortHookException("Spam content detected")
    return data

Priority -200: account_isolation_hook  (set account_id)
    ↓
Priority -150: created_by_hook         (set created_by/updated_by)
    ↓
Priority -100: timestamp_hook          (set timestamps)
    ↓
Priority 0+:   User hooks              (custom logic)
    ↓
Priority 100:  audit_capture_hook      (capture audit logs)

from snackbase.domain.entities.hook_context import AbortHookException

@app.state.hook.on_record_before_create("products", priority=50)
async def validate_product_price(event, data, context):
    """Ensure product price is positive."""
    if data and data.get("price", 0) <= 0:
        raise AbortHookException("Price must be greater than 0")
    return data

@app.state.hook.on_record_before_create("users", priority=50)
async def normalize_email(event, data, context):
    """Normalize email to lowercase."""
    if data and "email" in data:
        data["email"] = data["email"].lower().strip()
    return data

@app.state.hook.on_record_before_create("orders", priority=50)
async def calculate_total(event, data, context):
    """Calculate order total from line items."""
    if data and "items" in data:
        total = sum(item["price"] * item["quantity"] for item in data["items"])
        data["total"] = total
    return data

@app.state.hook.on_record_after_create("orders", priority=50)
async def send_order_notification(event, data, context):
    """Send email notification when order is created."""
    if data:
        await send_email(
            to=context.user.email,
            subject="Order Confirmation",
            body=f"Your order {data['id']} has been created."
        )
    return data

# Priority 200 - runs first
@app.state.hook.on_record_before_create("posts", priority=200)
async def hook_1(event, data, context):
    data["step"] = "1"
    return data

# Priority 100 - runs second
@app.state.hook.on_record_before_create("posts", priority=100)
async def hook_2(event, data, context):
    data["step"] = "2"
    return data

# Priority 0 (default) - runs last
@app.state.hook.on_record_before_create("posts")
async def hook_3(event, data, context):
    data["step"] = "3"
    return data

# Only for 'posts' collection
registry.register(
    event=HookEvent.ON_RECORD_BEFORE_CREATE,
    callback=my_hook,
    filters={"collection": "posts"}
)

# Default: errors are logged but don't stop execution
registry.register(
    event=HookEvent.ON_RECORD_AFTER_CREATE,
    callback=my_hook,
    stop_on_error=False  # Default
)

# Bad - does too much
async def mega_hook(event, data, context):
    validate_data(data)
    send_email(data)
    update_cache(data)
    return data

# Good - single responsibility
async def validate_data_hook(event, data, context):
    validate_data(data)
    return data

# Validation: High priority (runs early)
@app.state.hook.on_record_before_create("posts", priority=100)
async def validate(event, data, context):
    pass

# Enrichment: Medium priority
@app.state.hook.on_record_before_create("posts", priority=50)
async def enrich(event, data, context):
    pass

# Side effects: Low priority (runs late)
@app.state.hook.on_record_after_create("posts", priority=10)
async def notify(event, data, context):
    pass

@app.state.hook.on_record_after_create("posts")
async def send_notification(event, data, context):
    """Send notification, but don't fail if it errors."""
    try:
        await send_email(data)
    except Exception as e:
        logger.error("Failed to send notification", error=str(e))
    return data

class HookRegistry:
    def register(
        self,
        event: str,
        callback: Callable,
        filters: Optional[dict[str, Any]] = None,
        priority: int = 0,
        stop_on_error: bool = False,
        is_builtin: bool = False,
    ) -> str:
        """Register a hook."""

    async def trigger(
        self,
        event: str,
        data: Optional[dict[str, Any]] = None,
        context: Optional[HookContext] = None,
        filters: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        """Execute all hooks for an event."""

    def unregister(self, hook_id: str) -> bool:
        """Remove a registered hook."""

class AbortHookException(Exception):
    """Raise to abort an operation from a before hook."""
    def __init__(
        self,
        message: str,
        status_code: int = 400
    ):
        super().__init__(message)
        self.status_code = status_code