kicker 0.6.1

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

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
• Run jobs manually from UI
• Pause/resume scheduler and individual jobs
• Supports both sync and async functions
• Simple HTML interface

---

Installation

uv add kicker

---

Usage

Create a project and define your jobs:

# main.py
from kicker import Kicker

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(logger):
    logger.info("job_echo executed")


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

To see logs in the UI, declare a logger parameter — kicker injects it automatically. Jobs without it work fine but won't produce UI logs.

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.

# jobs/weekly_report.py
from kicker import Coworker

worker = Coworker()

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

# 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 see logs in a different output container, use the standatd extra keyword argument of the logger method you use (error in this example) with a output key and some string value as a new output container name:

from logging import Logger

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

or, cleaner with partial:

from functools import partial
from logging import Logger

@app.kick(second="*/5")
def sync_job(logger: Logger):
    log_error = partial(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 parameter, kicker also injects the scheduler object and a job_id of the current job object. This is useful for managing jobs — for example, we can change the next run time of the current job:

uv add --dev apscheduler

from datetime import datetime, timedelta
from functools import partial
from logging import Logger

from apscheduler.schedulers.base import BaseScheduler

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

---

Notes

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

---

License

MIT