Ashraful's Blog

Time to read: 5 min read

AsyncIO Task Runner (Coro Runner) - Simplifying Concurrent Python Task Management

AsyncIO Task Runner (Coro Runner) - Simplifying Concurrent Python Task Management

# Introduction

Managing asynchronous tasks in Python can be complex, especially when you need fine-grained control over concurrency limits and task scheduling. Enter Coro Runner (async-coro-runner), a lightweight Python utility that makes concurrent asynchronous task management straightforward and efficient.

In this post, we'll explore what makes Coro Runner unique, how to use it, and why it might be the perfect solution for your async workload management needs.

# What is Coro Runner?

Coro Runner is a Python library built on top of Python's native asyncio module that provides a simple yet powerful interface for managing concurrent asynchronous tasks. It's designed to simplify the execution of multiple async tasks with configurable concurrency limits, all within a single-threaded environment.

The library is particularly useful when you need to:

  • Execute multiple async tasks concurrently with controlled parallelism
  • Manage task queues with different priorities
  • Avoid overwhelming resources with too many concurrent operations
  • Schedule tasks dynamically from anywhere in your application

# Key Features

# 1. Configurable Concurrency

Define exactly how many tasks should run simultaneously, preventing resource exhaustion while maximizing throughput.

# 2. Simple API

Add tasks from anywhere in your codebase with a straightforward interface that doesn't require deep asyncio knowledge.

# 3. Worker Queue System

Multiple queues can be configured with their own priority levels, allowing sophisticated task management.

# 4. Built on asyncio

Leverages Python's battle-tested asyncio module (introduced in Python 3.4), ensuring stability and compatibility.

# Installation

Getting started is as simple as:

pip install coro-runner

# Quick Start Guide

Here's a basic example to get you started:

from coro_runner import CoroRunner

# Initialize the runner with a concurrency limit of 10
runner = CoroRunner(concurrency=10)

# Define an async task
async def process_data(item_id, **kwargs):
    # Your async processing logic here
    await some_async_operation(item_id)
    return f"Processed {item_id}"

# Add tasks from anywhere in your code
runner.add_task(process_data, args=[1], kwargs={"priority": "high"})
runner.add_task(process_data, args=[2], kwargs={"priority": "low"})
runner.add_task(process_data, args=[3], kwargs={"priority": "high"})

Important: Your task function must be an async function (defined with async def).

# Real-World Use Case: FastAPI Integration

One of the most powerful applications of Coro Runner is in web applications. Here's an example using FastAPI to handle background tasks:

from fastapi import FastAPI
from coro_runner import CoroRunner

app = FastAPI()

# Initialize runner once at startup
runner = CoroRunner(concurrency=10)

async def background_task(task_id: int):
    # Simulate some async work
    await asyncio.sleep(1)
    print(f"Task {task_id} completed")

@app.get("/fire-task")
async def trigger_tasks(count: int = 25):
    # Schedule multiple tasks
    for i in range(count):
        runner.add_task(background_task, args=[i])
    
    return {"message": f"Scheduled {count} tasks"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Why Choose Coro Runner?

# Compared to Raw asyncio

While Python's asyncio is powerful, managing task concurrency often requires boilerplate code. Coro Runner abstracts this complexity:

Without Coro Runner:

# Managing concurrency manually with asyncio
semaphore = asyncio.Semaphore(10)

async def limited_task(task_func, *args):
    async with semaphore:
        return await task_func(*args)

# Need to manage the semaphore for every task...

With Coro Runner:

# Simple and clean
runner = CoroRunner(concurrency=10)
runner.add_task(your_task, args=[1, 2, 3])

# Declare Once, Use Everywhere

A significant advantage is that you can initialize the runner once and call it from anywhere in your application:

# config.py
from coro_runner import CoroRunner
runner = CoroRunner(concurrency=20)

# module_a.py
from config import runner
runner.add_task(task_a, args=[1])

# module_b.py
from config import runner
runner.add_task(task_b, args=[2])

# Advanced Features (Coming Soon)

The Coro Runner roadmap includes exciting enhancements:

  • Monitoring Tool Integration: Real-time task monitoring and analytics
  • Low-Level API: Advanced features like callbacks, acknowledgments, and error handling
  • Robust Logging: Detailed execution tracking for debugging and optimization

# Use Cases

Coro Runner shines in scenarios such as:

  1. Web Scraping: Fetch multiple URLs concurrently while respecting rate limits
  2. Data Processing: Process large datasets with controlled parallelism
  3. API Integration: Call multiple external APIs without overwhelming them
  4. Background Jobs: Queue and execute background tasks in web applications
  5. Batch Operations: Perform bulk operations with controlled concurrency

# Example: Controlled Web Scraping

import aiohttp
from coro_runner import CoroRunner

runner = CoroRunner(concurrency=5)  # Only 5 concurrent requests

async def fetch_url(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# Schedule 100 URLs but only 5 will run at once
urls = [f"https://api.example.com/data/{i}" for i in range(100)]
for url in urls:
    runner.add_task(fetch_url, args=[url])

# Development and Contribution

Coro Runner is actively developed and welcomes contributions. The project uses Uv for dependency management and includes:

  • Comprehensive test suite with pytest
  • GitHub Actions for CI/CD
  • Docker support for isolated testing
  • Example applications demonstrating real-world usage

To contribute:

# Clone the repository
git clone https://github.com/iashraful/async-coro-runner.git
cd async-coro-runner

# Install dependencies
uv sync

# Run tests
uv run pytest -s

# Requirements

  • Python 3.12 or later
  • uv (for development)

# Conclusion

Coro Runner provides a clean, simple abstraction over asyncio's task management, making it easier to build high-performance concurrent applications in Python. Whether you're building web scrapers, API integrations, or background job processors, Coro Runner helps you manage concurrency without the complexity.

The library's philosophy is clear: provide powerful concurrency control with the simplest possible API. With its growing feature set and active development, Coro Runner is positioned to become an essential tool for Python developers working with asynchronous code.

Give it a try in your next project, and experience the simplicity of managed async task execution!

# Getting Help

Have questions or running into issues? Here's how to get help:

# Stay Updated

⭐ Star the repository on GitHub to stay updated with the latest features and improvements!

Have you used Coro Runner in your projects? Share your experience in the comments below!

Found a correction? Send me a PR! (opens new window)