Apscheduler

APScheduler Skill

Bookkeep uses APScheduler 3.x with AsyncIOScheduler for background job execution. Jobs are registered at startup via initialize_jobs(), persisted to JobSchedule table, and exposed through /api/jobs/ endpoints for runtime control.

Quick Start

Define a New Job

# backend/app/scheduler.py - Add to JOB_DEFINITIONS
JOB_DEFINITIONS = {
    "my_new_job": {
        "default_interval": 60 * 60,  # 1 hour
        "description": "Description for UI/logs",
        "type": "PROCESS",
    },
    # ... existing jobs
}

Implement the Job Function

# backend/app/tasks.py
async def my_new_job():
    """Background task description"""
    db: Session = SessionLocal()
    try:
        # Do work
        logger.info("my_new_job_complete", processed=count)
    except Exception as e:
        logger.error("my_new_job_error", error=str(e))
        db.rollback()
    finally:
        db.close()

Register in initialize_jobs()

# backend/app/scheduler.py - Add to job_functions dict
from app.tasks import my_new_job

job_functions = {
    "my_new_job": my_new_job,
    # ... existing mappings
}

Key Concepts

Common Patterns

Trigger Job Immediately

from app.scheduler import run_job_now
run_job_now("sync_from_booklore")  # Modifies next_run_time to now

Reschedule with New Interval

from app.scheduler import reschedule_job
reschedule_job("refresh_seed_data", 12 * 60 * 60)  # 12 hours

See Also

• patterns
• workflows

Related Skills

• See the python skill for async patterns and error handling
• See the sqlalchemy skill for JobSchedule model and database sessions
• See the fastapi skill for router endpoints in routers/jobs.py