Agent skill
apscheduler
Configures background job scheduling and task execution with APScheduler 3.x in FastAPI. Use when: adding new background jobs, modifying job intervals, implementing job state persistence, handling job events, or debugging scheduler issues.
Install this agent skill to your Project
npx add-skill https://github.com/majiayu000/claude-skill-registry/tree/main/skills/other/other/apscheduler-boludo00-bookkeep
SKILL.md
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
| Concept | Usage | Example |
|---|---|---|
AsyncIOScheduler |
Main scheduler for async jobs | scheduler = AsyncIOScheduler() |
IntervalTrigger |
Time-based job execution | IntervalTrigger(seconds=300) |
MemoryJobStore |
In-memory job storage | jobstores={'default': MemoryJobStore()} |
JobSchedule model |
Persist intervals/state to DB | See the sqlalchemy skill |
coalesce=True |
Combine missed executions | Prevents job pileup after downtime |
max_instances=1 |
Single concurrent execution | Prevents overlapping job runs |
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
JobSchedulemodel and database sessions - See the fastapi skill for router endpoints in
routers/jobs.py
Recommended Agent Skills
Expand your agent's capabilities with these related and highly-rated skills.
majiayu000/claude-skill-registry
agent-ops-spec
Manage specification documents in .agent/specs/. Use when user provides requirements, acceptance criteria, or feature descriptions that need to be tracked and validated against implementation.
majiayu000/claude-skill-registry
agent-ops-state
Maintain .agent state files. Use at session start, after meaningful steps, and before concluding: read/update constitution/memory/focus/issues/baseline consistently.
majiayu000/claude-skill-registry
agent-ops-spec
Manage specification documents in .agent/specs/. Use when user provides requirements, acceptance criteria, or feature descriptions that need to be tracked and validated against implementation.
majiayu000/claude-skill-registry
agent-ops-testing
Test strategy, execution, and coverage analysis. Use when designing tests, running test suites, or analyzing test results beyond baseline checks.
majiayu000/claude-skill-registry
agent-ops-testing
Test strategy, execution, and coverage analysis. Use when designing tests, running test suites, or analyzing test results beyond baseline checks.
majiayu000/claude-skill-registry
agent-ops-state
Maintain .agent state files. Use at session start, after meaningful steps, and before concluding: read/update constitution/memory/focus/issues/baseline consistently.
Didn't find tool you were looking for?