Common Patterns

Cookbook of recurring patterns in Kirak applications.

Ownership: Users Can Only See Their Own Records

Use an RLS condition to filter by the authenticated user’s ID:

{
  "posts": {
    "table": "posts",
    "schema": {
      "user_id": { "type": "integer" },
      "title":   { "type": "string" },
      "body":    { "type": "text" }
    },
    "access": {
      "fetch":  [{ "role": "user", "condition": "user_id = {user_id}" }],
      "create": [{ "role": "user", "condition": "user_id = {user_id}" }],
      "update": [{ "role": "user", "condition": "user_id = {user_id}" }],
      "delete": [{ "role": "user", "condition": "user_id = {user_id}" }]
    }
  }
}

On create, Kirak auto-injects user_id = <caller's JWT user_id> into the INSERT – no need to send it in the request body.

Admin Sees All, User Sees Own

{
  "orders": {
    "access": {
      "fetch": [
        { "role": "admin" },
        { "role": "user", "condition": "user_id = {user_id}" }
      ],
      "update": [
        { "role": "admin" },
        { "role": "user", "condition": "user_id = {user_id}" }
      ],
      "delete": [{ "role": "admin" }]
    }
  }
}

Admins see every row. Users see only rows they own. Delete is admin-only.

Tenant Isolation (Multi-Tenant SaaS)

Use organization_id as the RLS variable. It is not a token claim – the access token carries only sub, email and role for access rules, and Kirak cannot add claims to it – so it comes from an identity your own route sets:

{
  "invoices": {
    "access": {
      "fetch": [
        { "role": "user",  "condition": "organization_id = {organization_id}" },
        { "role": "admin", "condition": "organization_id = {organization_id}" }
      ],
      "create": [
        { "role": "user",  "condition": "organization_id = {organization_id}" },
        { "role": "admin", "condition": "organization_id = {organization_id}" }
      ]
    }
  }
}

On Kirak’s own routes (/api/invoices, GraphQL) {organization_id} resolves to NULL, so these rules return no rows and create nothing – they fail closed. Serve invoices from your own route: look up the caller’s organization, add it to the identity, and call Kirak inside it:

from fastapi import Request
from kirak.core.context import reset_user_context, set_user_context

SYSTEM_USER = {"role": "system", "user_id": None, "token": None}

async def caller_with_org(request: Request) -> dict:
    kirak = request.app.state.kirak
    caller = (await kirak.auth.get_current_user({"request": request})).get("data") or {}
    token = set_user_context(SYSTEM_USER)  # memberships needs a {"role": "system"} fetch rule
    try:
        membership = await kirak.fetch("memberships", {"user_id": caller["user_id"], "limit": 1})
    finally:
        reset_user_context(token)
    rows = membership.get("data") or []
    return {**caller, "organization_id": rows[0]["organization_id"] if rows else None}

@app.get("/org/invoices")
async def org_invoices(request: Request):
    token = set_user_context(await caller_with_org(request))
    try:
        return await request.app.state.kirak.fetch("invoices", {})
    finally:
        reset_user_context(token)

A caller without a membership gets organization_id = None, which matches no rows. Do not add the filter in a before_fetch hook instead: an async hook that raises or times out is skipped and the query runs unfiltered.

Soft Delete + Restore Pattern

Enable soft delete at the model level:

{
  "articles": {
    "table": "articles",
    "soft_delete": true,
    "schema": { "..." }
  }
}

DELETE /articles/delete?id=1 sets deleted_at – record is excluded from all fetch/search results.
PATCH /articles/restore?id=1 clears deleted_at.
DELETE /articles/destroy?id=1 physically removes the row (requires destroy permission).

Sending an Email After Registration

def on_kirak_ready(kirak):
    @kirak.auth.hook("after_register")
    async def send_welcome_email(result):
        if result.get("status") != "success":
            return result

        user = result["data"]["user"]
        try:
            await kirak.notifications.send_email({
                "to": user["email"],
                "subject": "Welcome to Acme",
                "text_body": f"Hi {user['first_name']}, thanks for signing up!",
            })
        except KirakException as e:
            # Log error but don't block registration
            print(f"Failed to send welcome email: {e.message}")
        return result

Webhook-Driven Fulfillment

Handle Stripe checkout.session.completed to fulfil an order:

def on_kirak_ready(kirak):
    @kirak.payments.hook("after_webhook")
    async def fulfil_order(result):
        if result["data"]["event"] != "payment_completed":
            return result
        order_id = result.get("metadata", {}).get("order_id")

        if order_id:
            await kirak.update("orders", {
                "where": {"id": order_id},
                "data": {"status": "paid"},
            })
        return result

Map metadata.order_id on the Stripe Checkout session when creating it so this hook can look up the order.

Scheduled Nightly Report

from datetime import date

def on_kirak_ready(kirak):
    @kirak.scheduler.cron("0 0 * * *", name="nightly_report")
    async def nightly_report(payload):
        # No request, no user: run as system. orders.access.fetch must list {"role": "system"}.
        token = set_user_context({"role": "system", "user_id": None, "token": None})
        try:
            result = await kirak.fetch("orders", {
                "status": "completed",
                "created_at__gte": date.today().isoformat(),
                "limit": 10000,
            })
        finally:
            reset_user_context(token)
        total = sum(r["amount"] for r in result["data"])

        try:
            await kirak.notifications.send_email({
                "to": "reports@acme.com",
                "subject": f"Daily revenue: ${total:.2f}",
                "text_body": f"{len(result['data'])} orders completed today.",
            })
        except KirakException as e:
            print(f"Failed to send nightly report: {e.message}")

0 0 * * * = midnight UTC every day. See Scheduler for cron expression format. set_user_context / reset_user_context come from kirak.core.context; a scheduled job has no identity of its own (Who a direct call runs as).

Background Image Processing

Upload to S3 and generate thumbnails in the background so the HTTP response returns immediately:

from fastapi import BackgroundTasks, Request, UploadFile
from kirak.core.context import reset_user_context, set_user_context

@app.post("/upload-avatar")
async def upload_avatar(request: Request, file: UploadFile, background_tasks: BackgroundTasks):
    kirak = request.app.state.kirak
    caller = (await kirak.auth.get_current_user({"request": request})).get("data") or {}
    file_bytes = await file.read()
    token = set_user_context(caller)  # storage refuses guests; stores under "{user_id}/"
    try:
        return await kirak.storage.upload_image({
            "file": file_bytes,
            "filename": file.filename,
            "path": "avatars/avatar.jpg",
            "thumbnails": [
                {"name": "sm", "width": 64,  "height": 64},
                {"name": "md", "width": 128, "height": 128},
            ],
            "background_tasks": background_tasks,  # thumbnails generated async
        })
    finally:
        reset_user_context(token)

Without background_tasks, thumbnails are generated synchronously before the response.

Aggregates and Joins in a Hook

Use GraphQL for queries the CRUD operations don’t express – sums, counts and averages grouped by a field, or related records in one query. kirak.graphql() goes through the same access blocks, row rules and field rules as every other operation:

TOTALS = """
query ($ids: [Int]) {
  line_items_aggregate(order_id: { in: $ids }, groupBy: "order_id") {
    order_id
    sum { amount }
  }
}
"""

def on_kirak_ready(kirak):
    @kirak.on("orders").hook("after_fetch")
    async def add_totals(result):
        ids = [r["id"] for r in result.get("data", [])]
        if not ids:
            return result

        totals = await kirak.graphql({"query": TOTALS, "variables": {"ids": ids}})
        rows = (totals.get("data") or {}).get("line_items_aggregate") or []
        total_map = {r["order_id"]: r["sum_amount"] for r in rows}
        for order in result["data"]:
            order["total"] = total_map.get(order["id"], 0)
        return result

Aggregates are checked against the model’s search rules, so line_items.access.search must allow the caller. See GraphQL for filters, grouping and relationships.

Avoid raw SQL (kirak.execute_query()): it skips the access block, row rules, field rules and hooks, so a mistake in it exposes data the model protects. The CRUD operations and GraphQL cover what a model holds.

Conditional Field Visibility

Use read_roles to hide sensitive fields from non-admin users:

{
  "users": {
    "schema": {
      "id":           { "type": "integer", "read_roles": ["admin", "user"] },
      "email":        { "type": "email",   "read_roles": ["admin", "user"] },
      "phone_number": { "type": "string",  "read_roles": ["admin"] },
      "internal_notes": { "type": "text", "read_roles": ["admin"] }
    }
  }
}

A user-role request to GET /users/fetch receives id and email only. phone_number and internal_notes are stripped by the field-level security layer before the response is sent.

Return extra data with every login, such as the user’s plan:

def on_kirak_ready(kirak):
    @kirak.auth.hook("after_login")
    async def add_subscription_plan(result):
        if result.get("status") != "success":
            return result

        user_id = result["data"]["user"]["user_id"]
        sub = await kirak.fetch("subscriptions", {
            "user_id": user_id,
            "status": "active",
            "limit": 1,
        })

        plan = "free"
        if sub["data"]:
            plan = sub["data"][0]["plan"]

        result["data"]["user"]["plan"] = plan
        return result

This changes the JSON the client receives, not the tokens: they are already signed when after_login runs, and a Kirak access token always carries exactly sub, email, role, jti, type, iat and exp. To act on the plan server-side, look it up in a before_* hook rather than reading it from the token.

Validating Input Before It Hits the Database

A sync before_* hook can reject an operation: raise a Kirak exception and the client receives the standard error envelope with that exception’s status code. The async def style used in most examples cannot reject; its errors are logged and skipped (see Hook Contract).

from kirak.core.exceptions import ValidationError

def on_kirak_ready(kirak):
    @kirak.on("bookings").hook("before_create")
    def check_dates(payload):
        data = payload.get("data", {})
        start, end = data.get("start_date"), data.get("end_date")
        if start and end and start >= end:
            raise ValidationError("end_date must be after start_date")
        return payload

Cross-field constraints that a single field’s required/enum can’t express (like start_date < end_date) belong in a sync hook like this one. Use an async def hook for side effects and payload enrichment that must not block the request. Hooks do not run inside the database transaction; see Transactions and Side Effects.

Public Read, Authenticated Write

{
  "articles": {
    "access": {
      "fetch":  [{ "role": "*" }],
      "search": [{ "role": "*" }],
      "count":  [{ "role": "*" }],
      "create": [{ "role": "editor" }, { "role": "admin" }],
      "update": [{ "role": "editor" }, { "role": "admin" }],
      "delete": [{ "role": "admin" }]
    }
  }
}

"role": "*" matches any user, including unauthenticated requests – this is how public endpoints work.

Health Check Endpoint

Add a /health route without going through the CRUD engine:

def on_kirak_ready(kirak):
    from fastapi import APIRouter
    router = APIRouter()

    @router.get("/health")
    async def health():
        return {"status": "ok", "version": "1.0.0"}

    kirak.app.include_router(router)