Jobify: An Asynchronous Task Scheduler Without External Infrastructure
Jobify lets you set up an asynchronous task scheduler in just a few lines of code. No message brokers needed, with SQLite as the default for persistence. Tasks are saved to disk and survive application restarts.
Basic example:
import asyncio
from jobify import Jobify
app = Jobify()
@app.task
async def send_notification(user_id: int) -> None:
print(f"Notify user {user_id}")
async def main() -> None:
async with app:
await send_notification.push(42)
print("Job scheduled in background")
asyncio.run(main())The push() method queues a task in the background without blocking. Jobify uses loop.call_at for precise timing instead of a polling loop, reducing idle load.
Scaling Challenges with Background Tasks
As the number of tasks grows, simple solutions fall short. The standard library's sched is only suitable for synchronous delayed calls without persistence and retries.
Linux cron is great for system scripts but doesn't integrate with your application. Celery requires brokers (RabbitMQ/Redis) and infrastructure. APScheduler is closer but limits task lifecycle management.
Jobify addresses this through:
- Full task lifecycle: statuses, retries, timeouts
- Middleware for execution and scheduling (outer middleware)
- Dependency injection via lifespan
- Automatic task deduplication
- Native asyncio support
FastAPI-Style API
FastAPI developers will quickly pick up Jobify. Familiar concepts include:
- Task decorators
@app.task - Lifespan for state initialization
- Middleware stack
- State injection via
INJECT
Example with result waiting:
import asyncio
from jobify import Jobify
app = Jobify()
@app.task
async def send_email(to: str, subject: str) -> None:
print(f"Sending email to {to}: {subject}")
async def main() -> None:
async with app:
job = await send_email.schedule(
to="[email protected]",
subject="Welcome!",
).delay(10)
await job.wait()
asyncio.run(main())The job object provides access to status, result, and cancellation.
Lifespan and Dependency Injection
Configuration via lifespan:
import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from jobify import INJECT, Jobify, State
@asynccontextmanager
async def lifespan(_: Jobify) -> AsyncIterator[dict[str, str]]:
yield {"sender": "[email protected]"}
app = Jobify(lifespan=lifespan)
@app.task
def send_email(to: str, subject: str, state: State = INJECT) -> None:
print(f"{state.sender} -> {to}: {subject}")Execution policies are set on the decorator:
@app.task(retry=3, timeout=30)
async def sync_user(user_id: int) -> None:
# ...Middleware: Managing Scheduling and Execution
Outer middleware trigger when a task is queued:
import asyncio
from jobify import Jobify, OuterContext
from jobify.middleware import BaseOuterMiddleware, CallNextOuter
class ScheduleLoggerMiddleware(BaseOuterMiddleware):
async def __call__(
self,
call_next: CallNextOuter,
context: OuterContext,
) -> asyncio.Handle:
print(f"Scheduling {context.job.id} with trigger: {context.trigger}")
return await call_next(context)
app = Jobify(outer_middleware=[ScheduleLoggerMiddleware()])Regular middleware extend execution (logging, metrics, retries).
Polling-Free Architecture
Jobify avoids while True: sleep(1): check_tasks(). Each task is immediately registered via loop.call_at for an exact time.
Advantages:
- Zero idle load
- Precise timing
- Doesn't block the event loop
Trade-off: reliance on monotonic time. If system time resets, tasks require recalculation.
Key Features
- Persistence: SQLite by default, tasks survive restarts
- Deduplication: automatic prevention of duplicates by ID
- Middleware stack: intervention in scheduling and execution
- FastAPI-like API: minimal learning curve for familiar developers
- Lightweight: no brokers, workers, or external dependencies
— Editorial Team
No comments yet.