Registry / workflow / aiojobs

aiojobs

JSON →
library1.4.0pypypi✓ verified 86d ago

aiojobs is an asyncio-based job scheduler for managing background tasks. It allows spawning and tracking long-running coroutines, handling their cancellation, and ensuring graceful shutdown. The current version is 1.4.0, and it follows an active release cadence with minor updates and bug fixes.

pip install aiojobs
INSTALL
IMPORT
SIG · AIOJOBS
A
aiojobs
workflowpythonv1.4.0
Install
1.6s avg
Import
200ms
Disk
16MB
Pass rate
10/ 10
Env Coverage10 / 10
glibc
3.93.13
musl
3.93.13
Install & Compatibility
Where this runs
tested against v1.4.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
musl
py 3.103.920 runs
installs and imports cleanly · install 0.0s · import 0.214s · 17.9MB
glibc
py 3.103.920 runs
installs and imports cleanly · install 1.6s · import 0.186s · 18MB
16MB installed
● package 16MB
Code
Verified usage

Verified import paths — ran on the pinned version, not inferred.

Scheduler
from aiojobs import Scheduler
import aiojobs; aiojobs.create_scheduler()
The `create_scheduler` function was removed in v1.1.0; `Scheduler` should be instantiated directly.
Job
from aiojobs import Job

This quickstart demonstrates how to create a `Scheduler` using the `async with` context (recommended for graceful shutdown since v1.3.0), spawn background tasks using `scheduler.spawn()`, and optionally wait for individual jobs to complete with `job.wait()`.

import asyncio from aiojobs import Scheduler async def worker(task_id: int): print(f"Worker {task_id}: Starting...") try: await asyncio.sleep(2) # Simulate work print(f"Worker {task_id}: Finished.") except asyncio.CancelledError: print(f"Worker {task_id}: Cancelled during sleep.") async def main(): # Use async with for graceful shutdown (introduced in v1.3.0) async with Scheduler() as scheduler: print("Scheduler created.") job1 = await scheduler.spawn(worker(1)) job2 = await scheduler.spawn(worker(2)) print(f"Spawned jobs: {scheduler.pending_count}") # Wait for job1 to complete (optional) await job1.wait() print(f"Job 1 status: {job1.closed}") print("Scheduler closed. All pending tasks should be done or cancelled.") if __name__ == "__main__": asyncio.run(main())
Debug
Known issues
breakingThe `aiojobs.create_scheduler()` function was removed. `Scheduler` objects must now be instantiated directly.
fix
Replace `await aiojobs.create_scheduler()` with `Scheduler()`.
affects: v1.1.0 and later
breakingPython 3.7 support was dropped.
fix
Upgrade your Python environment to 3.9 or higher (current minimum requirement for v1.4.0).
affects: v1.2.0 and later
gotchaTasks spawned by `Scheduler` may be abruptly cancelled during application shutdown if `Scheduler.wait_and_close()` is not called, or if the `Scheduler` is not used within an `async with` block.
fix
Use `async with Scheduler() as scheduler:` or manually call `await scheduler.wait_and_close()` before your application exits.
affects: All versions, specifically pre-v1.3.0 users without manual shutdown handling.
gotchaWhile `Scheduler` creation (v1.4.0+) no longer strictly requires a *running* event loop, tasks spawned by it still need an event loop to *execute*. Running `Scheduler` operations outside an `asyncio.run` context or a running event loop will lead to `RuntimeError` when tasks attempt to start.
fix
Always ensure your `aiojobs` operations are executed within an `async` function called by `asyncio.run()` or within an existing `asyncio` event loop.
affects: v1.4.0 and later (new behavior)
Errors
Common errors & fixes
AttributeError: module 'aiojobs' has no attribute 'create_scheduler'
Attempting to use the old API function `aiojobs.create_scheduler()` which was removed.
fix
Instantiate `Scheduler` directly: `scheduler = Scheduler()`.
Task was destroyed but it is pending!
The asyncio event loop was closed or the application exited while `aiojobs` tasks were still running or pending, without proper shutdown handling.
fix
Ensure graceful shutdown. Use `async with Scheduler() as scheduler:` or explicitly call `await scheduler.wait_and_close()` before closing your application.
TypeError: 'Scheduler' object is not awaitable
Attempting to `await` the `Scheduler` instantiation (e.g., `await Scheduler()`) which is not necessary and incorrect for direct instantiation.
fix
Instantiate `Scheduler` without `await`: `scheduler = Scheduler()`.
Upgrade
Version history
1.4.0latest on PyPI · released Apr 5, 2025
Audit
Dependencies
async-timeoutoptionalRequired for Python versions older than 3.11. Removed for Python 3.11+ in v1.2.0.
aiohttpoptionalRequired for aiohttp integration features (e.g., using aiohttp.web.AppKey).
Agent activity
55 hits · last 30 days
node
50
OpenAI (training)
1
Resources
aiojobs — pip install aiojobs · libregistry