Run Background Tasks with asyncio.create_task in Python
Create background tasks in an asyncio event loop with asyncio.create_task and run them concurrently using asyncio.gather.
Python code
27 linesimport asyncio
import time
async def background_worker(name, duration):
"""Simulates a long-running background task."""
print(f"{name} started at t={time.monotonic():.1f}")
await asyncio.sleep(duration)
print(f"{name} finished at t={time.monotonic():.1f}")
async def main():
print(f"Main starting at t={time.monotonic():.1f}")
# Create background tasks (not awaited immediately)
task1 = asyncio.create_task(background_worker("Worker A", 3))
task2 = asyncio.create_task(background_worker("Worker B", 1))
task3 = asyncio.create_task(background_worker("Worker C", 2))
# This runs immediately while background tasks proceed
print(f"Main doing something else at t={time.monotonic():.1f}")
await asyncio.sleep(0.5)
# Wait for all background tasks to complete
await asyncio.gather(task1, task2, task3)
print(f"All tasks done at t={time.monotonic():.1f}")
if __name__ == "__main__":
asyncio.run(main())Output
Main starting at t=0.0
Worker A started at t=0.0
Worker B started at t=0.0
Worker C started at t=0.0
Main doing something else at t=0.0
Worker B finished at t=1.0
Worker C finished at t=2.0
Worker A finished at t=3.0
All tasks done at t=3.0How it works
asyncio.create_task schedules the coroutine to run on the event loop immediately, but it does not block code that follows. The three workers start at nearly the same time because they are asynchronous. await asyncio.sleep(0.5) lets the main coroutine yield control so the workers can proceed. asyncio.gather waits for all tasks to complete before moving on. The event loop runs until asyncio.run(main()) finishes.
Common mistakes
- Forgetting to await asyncio.gather, causing warnings about never-awaited tasks.
- Calling the coroutine directly like `background_worker(...)` instead of wrapping with `asyncio.create_task`, which doesn't schedule it.
- Using `time.sleep` inside async functions, which blocks the event loop instead of `await asyncio.sleep`.
Variations
- Use `asyncio.wait` with `return_when=asyncio.FIRST_COMPLETED` to handle tasks as they finish.
- Apply `asyncio.TaskGroup` in Python 3.11+ for cleaner error handling.
Real-world use cases
- Kicking off multiple independent I/O operations, like parallel HTTP requests to different APIs.
- Running periodic background jobs (e.g., health checks or log rotation) while the main service stays responsive.
- Preloading data from several databases simultaneously before starting your application logic.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.