Horsies

PostgreSQL-backed background task queue and workflow engine for Python.

AI agents: see llms.txt for a complete sitemap and keyword index of these docs. If you are working from source, agent guidance skills are in horsies/.agents/skills/ (SKILL.md, tasks.md, workflows.md, configs.md).

Get Started Install Horsies and run your first task

Workflow API WorkflowSpec and WorkflowHandle signatures

View on GitHub Source code and contributions

Questions & Answers Design trade-offs, scaling, and common questions

Features

Postgres Based

Postgres based.

Type-safe Tasks

Tasks return typed results TaskResult[T, TaskError], allowing you to treat errors as values.

Real-time Dispatch

PostgreSQL LISTEN/NOTIFY for instant task delivery — no polling.

Workflow Engine

DAG based workflow definitions for orchestrating complex task pipelines.

Scheduled Tasks

Type-safe, human-readable schedule patterns with cron support.

Multi-queue Support

Priority-based queues with per-queue concurrency limits.

Syce Monitoring

TUI dashboard for monitoring workers, tasks, and workflows.

Quick Example

from horsies import Horsies, AppConfig, PostgresConfig, TaskResult, TaskError

app = Horsies(AppConfig(
    broker=PostgresConfig(
        database_url="postgresql+psycopg://user:pass@localhost:5432/mydb",
    ),
))

@app.task("add_numbers")
def add_numbers(a: int, b: int) -> TaskResult[int, TaskError]:
    return TaskResult(ok=a + b)

# Send and get result
handle = add_numbers.send(5, 3)
result = handle.get()
if result.is_ok():
    print(result.ok_value)  # 8