Tigrbl Cron

Delayed and recurring arbitrary task scheduling built from Tigrbl tables, a TigrblRouter, and an optional deployable TigrblApp.

Features

• The reusable router is the primary integration surface.
• The optional app mounts that router and includes Tigrbl diagnostics and docs.
• A task can be delayed once with run_at or repeated with cron_expression.
• Cron expressions are evaluated in-house using only the Python standard library.
• Recurrence is unlimited when valid_to is omitted.
• Any task can expire by setting valid_to.
• Any number of schedules can target the same registered handler.
• Every attempt is persisted as a TaskRun with its result or error.

Installation

uv add tigrbl-cron
# or: pip install tigrbl-cron

Tigrbl Cron targets Tigrbl 0.4.5 and its current public paths.

Router or app

Use the router inside a larger Tigrbl application:

from tigrbl import TigrblApp
from tigrbl_cron import build_router

cron_router = build_router(prefix="/cron")
app = TigrblApp(routers=[cron_router])

Or use the packaged application directly:

uvicorn tigrbl_cron.app:app

The module-level exports are also available as tigrbl_cron.router and tigrbl_cron.app.

Register arbitrary work

A handler is an ordinary synchronous or asynchronous callable. Its name is a lookup key; multiple ScheduledTask rows may use the same handler with different payloads and schedules.

from tigrbl_cron import register_task

@register_task("webhook.deliver")
async def deliver_webhook(*, task, session, scheduled_for, now):
    event = task.payload["event"]
    # Perform arbitrary application work here.
    return {"delivered": event, "at": now.isoformat()}

The handler receives:

• task: the persisted ScheduledTask, including its JSON payload;
• session: the active Tigrbl database session;
• scheduled_for: the schedule occurrence being processed;
• now: the worker sweep time.

Schedule delayed and repeated tasks

Scheduling uses Tigrbl's generated table operations. The REST routes below are created by the cron router.

import asyncio
from datetime import datetime, timedelta, timezone

import httpx

from tigrbl_cron import app

async def schedule() -> None:
    await app.initialize()
    now = datetime.now(timezone.utc)

    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport,
        base_url="http://cron",
    ) as client:
        # One delayed invocation.
        await client.post(
            "/cron/scheduledtask",
            json={
                "task_name": "webhook.deliver",
                "run_at": (now + timedelta(minutes=10)).isoformat(),
                "valid_from": now.isoformat(),
                "payload": {"event": "invoice.created"},
            },
        )

        # Unlimited recurrence: no valid_to and no recurrence count ceiling.
        await client.post(
            "/cron/scheduledtask",
            json={
                "task_name": "webhook.deliver",
                "cron_expression": "*/5 * * * *",
                "valid_from": now.isoformat(),
                "payload": {"event": "heartbeat"},
            },
        )

        # Repeated work that expires after one day.
        await client.post(
            "/cron/scheduledtask",
            json={
                "task_name": "webhook.deliver",
                "cron_expression": "0 * * * *",
                "valid_from": now.isoformat(),
                "valid_to": (now + timedelta(days=1)).isoformat(),
                "payload": {"event": "hourly.rollup"},
            },
        )

asyncio.run(schedule())

Exactly one of run_at and cron_expression is required. valid_to is optional for both schedule types.

Run a worker sweep

execute_due_tasks accepts either the cron router or the app that contains it. Call it from the worker loop or process manager of your choice:

import asyncio

from tigrbl_cron import execute_due_tasks, router

async def worker() -> None:
    await router.initialize()
    while True:
        await execute_due_tasks(router, limit=100)
        await asyncio.sleep(1)

asyncio.run(worker())

Each sweep executes at most one occurrence per schedule. When a worker missed multiple cron occurrences, the sweep coalesces them to the latest due time and the next sweep continues from there. A delayed task runs at most once. Expired tasks are left untouched.

Routes

The default router prefix is /cron:

• /cron/scheduledtask manages schedules.
• /cron/taskrun exposes execution history.

The optional app also exposes Tigrbl's standard documentation and diagnostics surfaces, including /docs, /openapi.json, and /system.

License

Licensed under the Apache License 2.0.