Skip to main content
Tasks and schedules keep background work inside the Capsule app. Use tasks for work triggered by a user or page action. Use schedules for recurring work.

​
Define a task

@app.task(retries=2, timeout=300)
async def sync_threads(owner_id: str):
    return {"synced": 25, "owner_id": owner_id}

​
Submit a task

@app.message()
async def handle(session: cpsl.Session, msg: cpsl.Message):
    handle = await sync_threads.submit(owner_id=session.user.owner_id)
    await session.show_task(handle, message="Sync started")

​
Add task options

@app.task(
    retries=3,
    timeout=600,
    lock="owner:{owner_id}",
    process=True,
)
async def rebuild_index(owner_id: str):
    ...
  • retries: retry transient failures
  • timeout: stop runaway work
  • lock: prevent duplicate work for the same key
  • process=True: run in a separate process for isolation

​
Schedule recurring work

@app.schedule("0 * * * *")
async def hourly_sync():
    await sync_threads.submit(owner_id="system")

​
Show task state on a page

@app.page("Tasks", icon="list-check")
def tasks_page():
    return cpsl.ui.Page([cpsl.ui.TaskBoard(title="Tasks")])