Metadata-Version: 2.3
Name: kicker
Version: 1.4.2
Summary: Lightweight job runner framework built on FastAPI and APScheduler with a simple web UI.
Author: Andrey Morozov
Author-email: Andrey Morozov <andrey@morozov.lv>
Requires-Dist: apscheduler>=3.11.2
Requires-Dist: dominate>=2.9.1
Requires-Dist: fastapi>=0.136.1
Requires-Dist: pydantic>=2.13.3
Requires-Dist: python-multipart>=0.0.28
Requires-Dist: uvicorn>=0.46.0
Requires-Python: >=3.14
Description-Content-Type: text/markdown

# kicker

Lightweight job runner built on FastAPI and APScheduler with a simple web UI.

## Features

* **FastAPI-style developer experience**
  * `app = Kicker()`
  * `@app.kick(...)`
  * standard `uvicorn module:app`
* Schedule jobs with APScheduler using decorators
* Run jobs manually from UI
* Pause/resume scheduler and individual jobs
* Supports both sync and async functions
* Simple HTML interface
* Use multiple log outputs
* Save execution state between job runs
* Supports multiple named instances of the same job
* Allows job-scoped setup/teardown using lifespan context manager
* Provides event handlers like `on_shutdown_requested` and `on_job_paused`

---

## Installation

```bash
uv add kicker
```

---

## Usage

Create a project and define your jobs:

```python
# main.py
from kicker import Kicker, JobContext

app = Kicker(logger_fmt="%(prefix)s[%(levelname)s] %(message)s at %(asctime)s")


@app.kick(day_of_week="mon-fri", hour="9-20/2", minute=0)
async def job_echo(ctx: JobContext):
    ctx.logger.info("job_echo executed")


# run the app
# uv run uvicorn package.module:app --reload
```

To see logs in the UI, use `ctx.logger`. Jobs that never log still run fine, but they will not produce UI output.

## Splitting jobs across files

Like FastAPI's `APIRouter`, `Coworker` lets you define jobs in separate files and include them in the main app — no circular imports, no shared `app` instance.

```python
# jobs/weekly_report.py
from kicker import Coworker, JobContext

worker = Coworker()


@worker.kick(hour=8, minute=0)
async def weekly_report(ctx: JobContext):
    ctx.logger.info("weekly_report executed")
```

```python
# main.py
from kicker import Kicker
from jobs.weekly_report import worker

app = Kicker()
app.include_coworker(worker)
```

## Adding multiple log outputs

It is possible to have multiple visual log output "containers". By default, the `default` output container is used.
To send logs to a different output container, pass the standard `extra` keyword argument to the logger method you use (`error` in this example)
with an `output` key and a string value as the new container name:

```python
from kicker import Kicker, JobContext

app = Kicker()


@app.kick(second="*/5")
def sync_job(ctx: JobContext):
    try:
        ...
    except:
        ctx.logger.error(
            "error in a sync job executed in the thread pool",
            extra={"output": "errors"},
        )
```

or, cleaner with `partial`:

```python
from functools import partial

from kicker import Kicker, JobContext

app = Kicker()


@app.kick(second="*/5")
def sync_job(ctx: JobContext):
    log_error = partial(ctx.logger.error, extra={"output": "errors"})
    try:
        ...
    except:
        log_error("error in a sync job executed in the thread pool")
```

## Getting access to the scheduler and the job

Along with the logger, kicker also injects the `scheduler` object and the `job_id` of the current job.
This is useful for managing jobs — for example, you can change the next run time of the current job:

```python
from datetime import datetime, timedelta
from functools import partial

from kicker import Kicker, JobContext

app = Kicker()


@app.kick(second="*/5")
def sync_job(ctx: JobContext):
    log_debug = partial(ctx.logger.debug, extra={"output": "debug"})
    slowdown_interval_minutes = 10
    try:
        ...
    except Exception as e:
        ctx.scheduler.modify_job(
            ctx.job_id,
            next_run_time=datetime.now() + timedelta(minutes=slowdown_interval_minutes),
        )
        log_debug(f"error: {e} - slowing down for {slowdown_interval_minutes} minutes")
```

## Save execution state between job runs

You can save arbitrary data in `ctx.storage` between job runs.

```python
from kicker import Coworker, JobContext

worker = Coworker()


@worker.kick(second="*/6")
async def very_important_job_with_runs_counter(ctx: JobContext):
    counter = ctx.storage.get("counter", 0)
    ctx.logger.info(f"important job executed ({counter}) times")
    ctx.storage["counter"] = counter + 1
```

## Run multiple named instances of the same job

It is possible to specify an instance name when including a coworker:

```python
from kicker import Kicker
from jobs.weekly_report import worker

app = Kicker()
app.include_coworker(worker)  # default instance
app.include_coworker(worker, instance="v2")
```

or just directly in the job decorator:

```python
from datetime import datetime, timedelta
from functools import partial

from kicker import Kicker, JobContext

app = Kicker()


def same_business_logic(ctx: JobContext):
    ctx.logger.info(
        f"sync job {ctx.instance} executed in the thread pool with a very long message",
        extra={"output": "sync_job"},
    )


@app.kick(second="*/5")
def sync_job1(ctx: JobContext):
    same_business_logic(ctx)


@app.kick(second="*/4", instance="v2")
def sync_job2(ctx: JobContext):
    same_business_logic(ctx)
```

## Visible job execution status with color intensity

You can report `failure` or `success` job execution status to the UI.
The status will be displayed with a green or red dot next to the job name.

The script below shows how to report job execution status with a color intensity.
Every error has a cooldown interval, which keeps the error visible for that period, but with reduced intensity.

```python
from datetime import datetime, UTC
from kicker import Coworker, JobContext

worker = Coworker()
FAILURE_COOLDOWN_MINUTES = 5
DEFAULT_ERROR_INTENSITY = 100


@worker.kick(second="*/30")
async def some_job(ctx: JobContext):
    try:
        ...
    except Exception:
        ctx.report_execution_status("failure", DEFAULT_ERROR_INTENSITY)
        ctx.storage["error_intensity"] = DEFAULT_ERROR_INTENSITY
        ctx.storage["last_error_time"] = datetime.now(UTC)
    else:
        if (
            datetime.now(UTC) - ctx.storage.get("last_error_time", datetime.min)
        ).total_seconds() > FAILURE_COOLDOWN_MINUTES * 60:
            ctx.report_execution_status("success")
        else:
            saved_intensity = ctx.storage.get(
                "error_intensity", DEFAULT_ERROR_INTENSITY
            )
            reduced_intensity = min(0, saved_intensity - 10)
            ctx.report_execution_status("failure", reduced_intensity)
            ctx.storage["error_intensity"] = reduced_intensity
```

## Worker setup/teardown using a lifespan context manager

If a job needs shared setup and teardown for all of its executions (for example a database engine),
you can pass a `lifespan` context manager to `@kick` or `include_coworker`. Setup runs once when the app starts,
before jobs run. Teardown runs on shutdown after in-flight jobs finish and the scheduler has stopped.
A job-level lifespan overrides a coworker-level one. Sync and async context managers are both supported.

```python
from contextlib import asynccontextmanager

from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import create_async_engine

from kicker import Coworker, JobContext

worker = Coworker()


@asynccontextmanager
async def db_lifespan(ctx: JobContext):
    engine = create_async_engine(
        "postgresql+asyncpg://user:password@localhost/db",
        echo=True,
    )
    ctx.storage["db_engine"] = engine
    ctx.logger.info("engine created")
    try:
        yield
    finally:
        await engine.dispose()
        del ctx.storage["db_engine"]


@worker.kick(second="*/6", lifespan=db_lifespan)
async def test_db_access(ctx: JobContext):
    engine = ctx.storage["db_engine"]
    try:
        async with engine.begin() as trn:
            await trn.execute(
                text(
                    "insert into table_name (column_name) select max(column_name)+1 from table_name"
                )
            )
        ctx.logger.info("inserted")
    except SQLAlchemyError as e:
        ctx.logger.error(f"Error occurred: {e}")
```

Various event handlers

```python
from kicker import Coworker, JobContext

worker = Coworker()


async def on_paused(ctx: JobContext):
    ctx.logger.info("I'm paused!")
    
async def on_system_shutdown(ctx: JobContext):
    ctx.logger.info("The system is about to be shut down!")

@worker.kick(second="*/6", 
             on_shutdown_requested=on_system_shutdown, 
             on_job_paused=on_paused)
async def test_on_pause(ctx: JobContext):
    ctx.logger.info("I'm working!")
```

---

## Notes

* Scheduler runs in-process (not distributed)
* Running with multiple workers will duplicate job execution
* Designed for simple internal tools and automation

---

## License

MIT
