Metadata-Version: 2.3
Name: orbi-sched
Version: 0.1.0
Summary: Lightweight decorator-based task scheduler for Python, built on APScheduler
Author: vgseven
Author-email: vgseven <ivgseven@gmail.com>
License: MIT
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Development Status :: 3 - Alpha
Requires-Dist: apscheduler>=3.11.2
Requires-Python: >=3.12
Project-URL: Homepage, https://github.com/vgseven/orbi-sched
Project-URL: Repository, https://github.com/vgseven/orbi-sched
Description-Content-Type: text/markdown

# orbi-sched

A lightweight task scheduler for Python, built on [APScheduler](https://apscheduler.readthedocs.io/). Decorate any function with `@orbi_task` to schedule it — no broker, no worker daemon, no configuration files required.

---

## Installation

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

**Requires:** Python 3.12+

---

## Quick Start

```python
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`

```python
@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:

```python
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

```python
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

```python
# 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](LICENSE).
