orbi-sched 0.1.0

Lightweight decorator-based task scheduler for Python, built on APScheduler

orbi-sched

A lightweight task scheduler for Python, built on APScheduler. Decorate any function with @orbi_task to schedule it — no broker, no worker daemon, no configuration files required.

---

Installation

pip install orbi-sched
# or with uv
uv add orbi-sched

Requires: Python 3.12+

---

Quick Start

from orbi_task.scheduler import orbi_task, get_scheduler

# Run once at a specific datetime
@orbi_task(trigger="date", time="2026-04-01 09:00:00")
def send_report():
    print("Sending daily report...")

# Run every weekday at 08:30
@orbi_task(trigger="cron", day_of_week="mon-fri", hour=8, minute=30)
def morning_sync():
    print("Syncing data...")

# Keep the process alive
import time
while True:
    time.sleep(5)

---

API Reference

@orbi_task

@orbi_task(
    trigger: Literal["date", "cron"] = "cron",
    *,
    time: str | None = None,        # "YYYY-MM-DD HH:MM:SS" — date trigger only
    day_of_week: str | None = None, # e.g. "mon", "mon-fri", "0,2,4"
    hour: int | None = None,
    minute: int | None = None,
    **kwargs,                        # passed directly to APScheduler trigger
)

Registers the decorated function as a scheduled job. The function runs in a background thread managed by APScheduler.

Parameter	Trigger	Description
trigger	both	"date" for one-shot, "cron" for recurring
time	date only	Exact datetime string to run the job
day_of_week	cron	Days to run (APScheduler cron syntax)
hour	cron	Hour of day (0–23)
minute	cron	Minute of hour (0–59)
**kwargs	both	Any additional APScheduler trigger arguments

The job ID is automatically set to orbi_{function_name}. Re-decorating with the same function name replaces the existing job.

Note: The wrapper returned by the decorator is a no-op. Jobs are registered at decoration time and run in the background scheduler — calling the decorated function directly does nothing.

---

get_scheduler() -> BackgroundScheduler

Returns the global singleton BackgroundScheduler instance, initializing it on first call. Use this to inspect, pause, or shut down the scheduler:

sched = get_scheduler()
sched.print_jobs()          # list all registered jobs
sched.shutdown(wait=True)   # graceful shutdown

Defaults set on the scheduler:

• misfire_grace_time: 60 seconds — a job that missed its window by less than 60s will still run
• coalesce: True — if multiple runs were missed, only one catch-up run fires
• timezone: Asia/Kolkata (IST)

---

Execution Logs

The scheduler emits structured console output on every job event:

🔥 Job 'orbi_morning_sync' executed at 2026-04-01 08:30:00+05:30 (next: 2026-04-02 08:30:00+05:30)
💥 Job 'orbi_send_report' ERROR: <exception message>

---

Graceful Shutdown

import signal, sys
from orbi_task.scheduler import get_scheduler

def shutdown(sig, frame):
    get_scheduler().shutdown(wait=True)
    sys.exit(0)

signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)

---

Trigger Examples

# One-shot: run at a specific time
@orbi_task(trigger="date", time="2026-06-15 14:00:00")
def one_time_task(): ...

# Every day at midnight
@orbi_task(trigger="cron", hour=0, minute=0)
def nightly_cleanup(): ...

# Every Monday and Wednesday at 10:15
@orbi_task(trigger="cron", day_of_week="mon,wed", hour=10, minute=15)
def twice_weekly(): ...

---

Dependencies

Package	Version
apscheduler	>=3.11.2

---

License

MIT — see LICENSE.