tickforge

An advanced task scheduler for Python with persistence, native async support, retries, timeouts, misfire handling and a batteries-included CLI.

Features

• Three trigger types — IntervalTrigger, CronTrigger (with timezone support), DateTrigger
• Durable jobs — SQLite, JSON file, or in-memory stores behind one interface
• Async first — async def jobs are awaited; sync jobs run in a thread pool
• Reliability — per-job retries, retry backoff delay, execution timeouts, misfire grace windows
• Concurrency control — global semaphore plus per-job overlap protection
• Run history — every execution recorded with duration, attempts, result and traceback
• Event hooks — subscribe to job and scheduler lifecycle events
• Structured logging — human-readable or JSON, with job context on every record
• CLI — register, inspect, preview, run and serve jobs without writing a driver script

Installation

pip install tickforge

From source:

git clone https://github.com/example/tickforge
cd tickforge
pip install -e ".[dev]"

Requires Python 3.8 or newer. Dependencies: click, python-dateutil.

Quick start

Async

import asyncio
from tickforge import AsyncScheduler, CronTrigger, IntervalTrigger, SQLiteJobStore

async def heartbeat():
    print("alive")

def collect_metrics(source: str):
    print("collecting from", source)

async def main():
    scheduler = AsyncScheduler(store=SQLiteJobStore("jobs.db"))
    await scheduler.add_job(heartbeat, IntervalTrigger(seconds=30), name="heartbeat")
    await scheduler.add_job(
        collect_metrics,
        CronTrigger.from_string("*/5 * * * *", timezone="Europe/Paris"),
        args=["edge-01"],
        max_retries=3,
        retry_delay=10,
        timeout=60,
        name="metrics",
    )
    await scheduler.run_forever()

asyncio.run(main())

Synchronous

from tickforge import Scheduler, IntervalTrigger, MemoryJobStore

def report():
    print("report generated")

with Scheduler(store=MemoryJobStore()) as scheduler:
    scheduler.add_job(report, IntervalTrigger(minutes=15), name="report")
    input("press enter to stop\n")

Triggers

Cron fields are minute hour day month day_of_week, supporting *, a-b, a,b, */n, a-b/n, month and weekday names, and the macros @hourly, @daily, @weekly, @monthly, @yearly.

day_of_week uses Python semantics: 0 is Monday through 6 is Sunday. Names (mon, fri) and the legacy 7 for Sunday are also accepted. When both day and day_of_week are restricted, classic cron OR semantics apply.

Stores

from tickforge import MemoryJobStore, JSONFileJobStore, SQLiteJobStore, create_store

MemoryJobStore()                  # volatile, ideal for tests
JSONFileJobStore("jobs.json")     # human-readable, atomic writes
SQLiteJobStore("jobs.db")         # durable, WAL, safe across threads and processes

create_store("sqlite:///var/lib/tickforge/jobs.db")
create_store("json:///tmp/jobs.json")
create_store("memory://")

Because jobs are persisted as module:callable references, the target must be importable from the process running the scheduler. Lambdas and locally defined functions are rejected at registration time.

Reliability options

await scheduler.add_job(
    flaky_task,
    IntervalTrigger(minutes=5),
    max_retries=3,            # four attempts total
    retry_delay=10,           # seconds between attempts
    timeout=120,              # abort a single attempt after two minutes
    misfire_grace_time=300,   # drop slots older than five minutes
    coalesce=True,            # collapse a missed backlog into one run
    allow_concurrent=False,   # skip a slot if the previous run is still going
)

Events

from tickforge import EventType

def on_error(event):
    print("job failed:", event.job.name, event.run.error)

scheduler.add_listener(on_error, EventType.JOB_ERROR)
scheduler.add_listener(lambda e: print(e.to_dict()))   # all events

Available types: SCHEDULER_STARTED, SCHEDULER_STOPPED, SCHEDULER_PAUSED, SCHEDULER_RESUMED, JOB_ADDED, JOB_REMOVED, JOB_MODIFIED, JOB_SUBMITTED, JOB_EXECUTED, JOB_ERROR, JOB_RETRY, JOB_MISSED, JOB_SKIPPED, JOB_FINISHED.

Logging

from tickforge import configure_logging

configure_logging(level="DEBUG", fmt="json", log_file="/var/log/tickforge.log")

Records emitted during a run carry job_id, run_id and job_name automatically.

CLI

# register jobs
tickforge add myapp.tasks:cleanup --interval 1h --name nightly-cleanup
tickforge add myapp.tasks:report --cron "0 9 * * mon-fri" --timezone Europe/Paris
tickforge add myapp.tasks:ping --at "2026-06-01 12:00" --retries 2 --timeout 30

# inspect
tickforge list
tickforge show nightly-cleanup
tickforge next nightly-cleanup --count 10
tickforge history --limit 20

# control
tickforge pause nightly-cleanup
tickforge resume nightly-cleanup
tickforge run-now nightly-cleanup
tickforge remove nightly-cleanup --yes

# serve
tickforge start --poll 1 --concurrency 20

Global options: --db (path or store URI, also read from TICKFORGE_DB), -v/-vv, -q, --log-format text|json, --log-file.

The default store is ~/.tickforge/jobs.db.

Running as a service

[Unit]
Description=tickforge scheduler
After=network.target

[Service]
Environment=TICKFORGE_DB=/var/lib/tickforge/jobs.db
ExecStart=/usr/local/bin/tickforge --log-format json start
Restart=always
User=tickforge

[Install]
WantedBy=multi-user.target

Development

pip install -e ".[dev]"
pytest
ruff check src
black src
mypy src

License

MIT — see LICENSE.