Lifecycle Hooks¶

Hooks let you inject custom logic at every stage of a resource's lifecycle — validate input before create, trigger a side-effect after update, block a delete under certain conditions, and so on. They run inside the same request as the CRUD operation, so they see the same user, the same transaction intent, and the same data the default handler sees.

Hooks are the extension point of choice when a Resource's default behaviour is almost right. When you need to replace the handler entirely, override the method on a Resource subclass instead.

Hook types¶

There are seven hook types, covering every CRUD lifecycle event:

Hook	Fires	Can mutate `ctx.data`	Typical use
`BEFORE_CREATE`	Just before the document is inserted.	Yes	Validate, enrich, derive fields.
`AFTER_CREATE`	After a successful insert.	No	Send notifications, emit events, queue jobs.
`BEFORE_UPDATE`	Just before the document is updated.	Yes	Validate transitions, sanitize, lock fields.
`AFTER_UPDATE`	After a successful update.	No	Cascade, notify, re-compute derived data.
`BEFORE_DELETE`	Just before the document is deleted.	No	Enforce business rules, block deletes.
`AFTER_DELETE`	After a successful (soft) delete.	No	Clean up related state, notify.
`AFTER_FETCH`	For every item returned by a list or get.	No	Strip or decorate fields per request.

BEFORE_CREATE and BEFORE_UPDATE hooks can return a dict to replace ctx.data. Every other hook type should return None — its return value is ignored.

Hook context¶

Every hook receives a single HookContext argument:

class HookContext(BaseModel):
    resource_name: str         # Prefix of the Resource, e.g. "products"
    user: Any = None           # The authenticated user, or None
    item: Any = None           # The document (set for AFTER_* and BEFORE_DELETE)
    data: Optional[dict]       # Input payload (set for BEFORE_CREATE / BEFORE_UPDATE)
    previous: Optional[dict]   # Pre-update snapshot (set for AFTER_UPDATE)
    snapshot: Optional[dict]   # Pre-delete snapshot (set for AFTER_DELETE)
    meta: Optional[dict]       # Free-form metadata (e.g. bulk_operation_id)

Which fields are populated depends on the hook type. A quick reference:

Hook	`item`	`data`	`previous`	`snapshot`
`BEFORE_CREATE`	—	✓	—	—
`AFTER_CREATE`	✓	—	—	—
`BEFORE_UPDATE`	✓	✓	—	—
`AFTER_UPDATE`	✓	—	✓	—
`BEFORE_DELETE`	✓	—	—	—
`AFTER_DELETE`	✓	—	—	✓
`AFTER_FETCH`	✓	—	—	—

Registering a hook¶

Every Resource has its own hooks registry. Register a hook by calling resource.hooks.register(...) before you pass the resource to register_resource:

from craft_easy import create_base_app
from craft_easy.core.crud import Resource
from craft_easy.core.hooks import HookContext, HookType
from my_project.models.product import Product

def create_app():
    app = create_base_app()

    resource = Resource(Product, path="/products")

    async def block_negative_prices(ctx: HookContext) -> dict | None:
        if ctx.data and ctx.data.get("price", 0) < 0:
            raise ValueError("Price must be non-negative")
        return None

    resource.hooks.register(HookType.BEFORE_CREATE, block_negative_prices)
    resource.hooks.register(HookType.BEFORE_UPDATE, block_negative_prices)

    app.register_resource(resource)
    return app

Mutating input in BEFORE hooks¶

If a BEFORE_CREATE or BEFORE_UPDATE hook returns a dict, that dict replaces ctx.data. This is how you enrich or sanitize input before it reaches the database:

async def normalize_sku(ctx: HookContext) -> dict | None:
    if not ctx.data:
        return None
    sku = ctx.data.get("sku")
    if isinstance(sku, str):
        ctx.data["sku"] = sku.strip().upper()
    return ctx.data

resource.hooks.register(HookType.BEFORE_CREATE, normalize_sku)
resource.hooks.register(HookType.BEFORE_UPDATE, normalize_sku)

Returning None leaves ctx.data untouched. Hooks in the chain always see the most recent version of ctx.data, so you can compose several before-hooks that each tweak one field.

Side-effects in AFTER hooks¶

AFTER hooks are where you trigger notifications, emit events, queue jobs, or propagate to other systems:

async def notify_low_stock(ctx: HookContext) -> None:
    if ctx.item and ctx.item.quantity < 10:
        await notifications.send(
            to_role="warehouse",
            template="low_stock",
            data={"product_id": str(ctx.item.id), "name": ctx.item.name},
        )

resource.hooks.register(HookType.AFTER_UPDATE, notify_low_stock)

Hooks run inside the request. An exception raised from an AFTER hook will bubble up and turn the response into a 5xx. If your side-effect is tolerant of failure, catch the exception and log it; if the side-effect is essential, let it fail loud.

Blocking deletes¶

A BEFORE_DELETE hook that raises an exception aborts the delete:

from craft_easy.core.errors.exceptions import ConflictError

async def block_delete_with_orders(ctx: HookContext) -> None:
    if ctx.item and await Order.find({"product_id": ctx.item.id}).count() > 0:
        raise ConflictError("Cannot delete product with associated orders")

resource.hooks.register(HookType.BEFORE_DELETE, block_delete_with_orders)

For reference-based deletion guards, configure CascadeConfig.delete = {"product_id": "deny"} on the Order model instead — cascade rules are declarative and apply across every resource that references the target. See Cascade Operations.

Execution order¶

Hooks for a given type execute in registration order. The Resource registers a set of automatic hooks for audit logging and cascade operations before your hooks get a chance to run. For each CRUD lifecycle event, the order is:

Auto-hooks (cascade population, audit log entries, cascade delete/update).
Your registered hooks in the order you registered them.

If two before-hooks both return dicts, the second one operates on the output of the first.

The `AFTER_FETCH` hook¶

AFTER_FETCH fires for every item returned by a list or get endpoint — before attribute-level access filtering runs. Use it to strip or decorate fields per-request:

async def redact_for_guest(ctx: HookContext) -> None:
    if ctx.item and ctx.user is None:
        ctx.item.internal_notes = None

resource.hooks.register(HookType.AFTER_FETCH, redact_for_guest)

Because it runs once per document, avoid expensive I/O here; do that in AFTER_UPDATE/AFTER_CREATE and cache the result on the document.

Testing hooks¶

Hooks are plain async functions — you can unit-test them in isolation by constructing a HookContext directly:

import pytest
from craft_easy.core.hooks import HookContext, HookType

@pytest.mark.asyncio
async def test_block_negative_prices():
    ctx = HookContext(resource_name="products", data={"name": "x", "price": -1})
    with pytest.raises(ValueError):
        await block_negative_prices(ctx)

Source¶

craft_easy/core/hooks/types.py — HookType enum, HookContext schema.
craft_easy/core/hooks/registry.py — HookRegistry and the execution loop.
craft_easy/core/crud/auto_hooks.py — automatic cascade and audit hooks registered for every Resource.