manutej/luxor-claude-marketplace/asyncio-concurrency-patterns

Claude Code and agent skill by manutej/luxor-claude-marketplace

Trust tier 2
  • community
LLM review pending
Source check pending

Checking that the SKILL.md file is still in the repository.

Complete guide for asyncio concurrency patterns including event loops, coroutines, tasks, futures, async context managers, and performance optimization

Install

npx vskill@latest install manutej/luxor-claude-marketplace/asyncio-concurrency-patterns

SKILL.md

Asyncio Concurrency Patterns

A comprehensive skill for mastering Python's asyncio library and concurrent programming patterns. This skill covers event loops, coroutines, tasks, futures, synchronization primitives, async context managers, and production-ready patterns for building high-performance asynchronous applications.

When to Use This Skill

Use this skill when:

  • Building I/O-bound applications that need to handle many concurrent operations
  • Creating web servers, API clients, or websocket applications
  • Implementing real-time systems with event-driven architecture
  • Optimizing application performance with concurrent request handling
  • Managing multiple async operations with proper coordination and error handling
  • Building background task processors or job queues
  • Implementing async database operations and connection pooling
  • Creating chat applications, real-time dashboards, or notification systems
  • Handling parallel HTTP requests efficiently
  • Managing websocket connections with multiple event sources
  • Building microservices with async communication patterns
  • Optimizing resource utilization in network applications

Core Concepts

What is Asyncio?

Asyncio is Python's built-in library for writing concurrent code using the async/await syntax. It provides:

  • Event Loop: The core of asyncio that schedules and runs asynchronous tasks
  • Coroutines: Functions defined with async def that can be paused and resumed
  • Tasks: Scheduled coroutines that run concurrently
  • Futures: Low-level objects representing results of async operations
  • Synchronization Primitives: Locks, semaphores, events for coordination
Event Loop Fundamentals

The event loop is the central execution mechanism in asyncio:

python
import asyncio
# Get or create an event loop
loop = asyncio.get_event_loop()
# Run a coroutine until complete
loop.run_until_complete(my_coroutine())
# Modern approach (Python 3.7+)
asyncio.run(my_coroutine())

Key Event Loop Concepts:

  1. Single-threaded concurrency: One thread, many tasks
  2. Cooperative multitasking: Tasks yield control voluntarily
  3. I/O multiplexing: Efficient handling of many I/O operations
  4. Non-blocking operations: Don't wait for I/O, do other work
Coroutines vs Functions

Regular Function:

python
def fetch_data():
    # Blocks until complete
    return requests.get('http://api.example.com')

Coroutine:

python
async def fetch_data():
    # Yields control while waiting
    async with aiohttp.ClientSession() as session:
        async with session.get('http://api.example.com') as resp:
            return await resp.text()
Tasks and Futures

Tasks wrap coroutines and schedule them on the event loop:

python
# Create a task
task = asyncio.create_task(my_coroutine())
# Task runs in background
# ... do other work ...
# Wait for result
result = await task

Futures represent eventual results:

python
# Low-level future (rarely used directly)
future = asyncio.Future()
# Set result
future.set_result(42)
# Get result
result = await future
Async Context Managers

Manage resources with async setup/teardown:

python
class AsyncResource:
    async def __aenter__(self):
        # Async setup
        await self.connect()
        return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Async cleanup
        await self.disconnect()
# Usage
async with AsyncResource() as resource:
    await resource.do_work()

Concurrency Patterns

Pattern 1: Gather - Concurrent Execution

Run multiple coroutines concurrently and wait for all to complete:

python
import asyncio
import aiohttp
async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()
async def main():
    async with aiohttp.ClientSession() as session:
        # Run all fetches concurrently
        results = await asyncio.gather(
            fetch(session, 'http://python.org'),
            fetch(session, 'http://docs.python.org'),
            fetch(session, 'http://pypi.org')
        )
        return results
# Results is a list in the same order as inputs
results = asyncio.run(main())

When to use:

  • Need all results
  • Order matters
  • Want to fail fast on first exception (default)
  • Can handle partial results with return_exceptions=True
Pattern 2: Wait - Flexible Waiting

More control over how to wait for multiple tasks:

python
import asyncio
async def task_a():
    await asyncio.sleep(2)
    return 'A'
async def task_b():
    await asyncio.sleep(1)
    return 'B'
async def main():
    tasks = [
        asyncio.create_task(task_a()),
        asyncio.create_task(task_b())
    ]
    # Wait for first to complete
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )
    # Get first result
    first_result = done.pop().result()
    # Cancel remaining
    for task in pending:
        task.cancel()
    return first_result
result = asyncio.run(main())  # Returns 'B' after 1 second

Wait strategies:

  • FIRST_COMPLETED: Return when first task finishes
  • FIRST_EXCEPTION: Return when first task raises exception
  • ALL_COMPLETED: Wait for all tasks (default)
Pattern 3: Semaphore - Limit Concurrency

Control maximum number of concurrent operations:

python
import asyncio
import aiohttp
async def fetch_with_limit(session, url, semaphore):
    async with semaphore:
        # Only N requests run concurrently
        async with session.get(url) as resp:
            return await resp.text()
async def main():
    # Limit to 5 concurrent requests
    semaphore = asyncio.Semaphore(5)
    urls = [f'http://api.example.com/item/{i}' for i in range(100)]
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(session, url, semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
    return results
asyncio.run(main())

When to use:

  • Rate limiting API requests
  • Controlling database connection usage
  • Preventing resource exhaustion
  • Respecting external service limits
Pattern 4: Lock - Mutual Exclusion

Ensure only one coroutine accesses a resource at a time:

python
import asyncio
class SharedCounter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()
    async def increment(self):
        async with self.lock:
            # Critical section - only one coroutine at a time
            current = self.value
            await asyncio.sleep(0)  # Simulate async work
            self.value = current + 1
async def worker(counter):
    for _ in range(100):
        await counter.increment()
async def main():
    counter = SharedCounter()
    # Run 10 workers concurrently
    await asyncio.gather(*[worker(counter) for _ in range(10)])
    print(f"Final count: {counter.value}")  # Always 1000
asyncio.run(main())
Pattern 5: Event - Signaling

Coordinate multiple coroutines with events:

python
import asyncio
async def waiter(event, name):
    print(f'{name} waiting for event')
    await event.wait()
    print(f'{name} received event')
async def setter(event):
    await asyncio.sleep(2)
    print('Setting event')
    event.set()
async def main():
    event = asyncio.Event()
    # Multiple waiters
    await asyncio.gather(
        waiter(event, 'Waiter 1'),
        waiter(event, 'Waiter 2'),
        waiter(event, 'Waiter 3'),
        setter(event)
    )
asyncio.run(main())
Pattern 6: Queue - Producer/Consumer

Coordinate work between producers and consumers:

python
import asyncio
async def producer(queue, n):
    for i in range(n):
        await asyncio.sleep(0.1)
        await queue.put(f'item-{i}')
        print(f'Produced item-{i}')
    # Signal completion
    await queue.put(None)
async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            # Propagate sentinel to other consumers
            await queue.put(None)
            break
        print(f'{name} processing {item}')
        await asyncio.sleep(0.2)
        queue.task_done()
async def main():
    queue = asyncio.Queue()
    # Start producer and consumers
    await asyncio.gather(
        producer(queue, 10),
        consumer(queue, 'Consumer-1'),
        consumer(queue, 'Consumer-2'),
        consumer(queue, 'Consumer-3')
    )
asyncio.run(main())

Task Management

Creating Tasks

Basic Task Creation:

python
import asyncio
async def background_task():
    await asyncio.sleep(10)
    return 'Done'
async def main():
    # Create task - starts running immediately
    task = asyncio.create_task(background_task())
    # Do other work while task runs
    await asyncio.sleep(1)
    # Wait for result
    result = await task
    return result
asyncio.run(main())

Named Tasks (Python 3.8+):

python
task = asyncio.create_task(
    background_task(),
    name='my-background-task'
)
print(task.get_name())  # 'my-background-task'
Task Cancellation

Graceful Cancellation:

python
import asyncio
async def long_running_task():
    try:
        while True:
            await asyncio.sleep(1)
            print('Working...')
    except asyncio.CancelledError:
        print('Task cancelled, cleaning up...')
        # Cleanup logic
        raise  # Re-raise to mark as cancelled
async def main():
    task = asyncio.create_task(long_running_task())
    # Let it run for 3 seconds
    await asyncio.sleep(3)
    # Request cancellation
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('Task was cancelled')
asyncio.run(main())

Cancellation with Context Manager:

python
import asyncio
from contextlib import suppress
async def run_with_timeout():
    task = asyncio.create_task(long_running_task())
    try:
        # Wait with timeout
        await asyncio.wait_for(task, timeout=5.0)
    except asyncio.TimeoutError:
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task
Exception Handling in Tasks

Gather with Exception Handling:

python
import asyncio
async def failing_task(n):
    await asyncio.sleep(n)
    raise ValueError(f'Task {n} failed')
async def successful_task(n):
    await asyncio.sleep(n)
    return f'Task {n} succeeded'
async def main():
    # return_exceptions=True: Returns exceptions instead of raising
    results = await asyncio.gather(
        successful_task(1),
        failing_task(2),
        successful_task(3),
        return_exceptions=True
    )
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f'Task {i} failed: {result}')
        else:
            print(f'Task {i} result: {result}')
asyncio.run(main())

Task Exception Retrieval:

python
import asyncio
async def main():
    task = asyncio.create_task(failing_task(1))
    # Wait for task
    await asyncio.sleep(2)
    # Check if task failed
    if task.done() and task.exception():
        print(f'Task failed with: {task.exception()}')
asyncio.run(main())

Event Loop Management

Event Loop Policies

Default Event Loop:

python
import asyncio
async def main():
    # Get running loop
    loop = asyncio.get_running_loop()
    print(f'Loop: {loop}')
asyncio.run(main())

Custom Event Loop:

python
import asyncio
async def main():
    pass
# Create new event loop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
    loop.run_until_complete(main())
finally:
    loop.close()

Event Loop Best Practices:

  1. Use `asyncio.run()` for simple programs (Python 3.7+)
  2. Avoid creating ClientSession outside event loop
  3. Always close loops when done
  4. Don't call blocking functions in event loop

This is the start of the file. Read the full SKILL.md on GitHub.

Security

  • Security scanPassed, 100 of 100Pattern checks on the skill's files.
  • Model reviewNot runA model review is scheduled for the next rescan.

See the full security report

Quality evaluation

  • Task evaluationNot runNo evaluation yet. Run one in Skill Studio.

See evaluation results

Source

Popularity

Repository stars
62
Repository forks
14
7-day trend score
21

Stars and forks are for the whole GitHub repository, not this skill alone. Refreshed 25 Apr 2026, 6 months ago.

Works with

Any agent vskill installs to: 56 targets in all. These 39 have a page on this site.

  • Claude Code
  • Cursor
  • GitHub Copilot
  • Windsurf
  • Codex
  • Gemini CLI
  • Cline
  • Amp
  • Roo Code
  • Goose
  • Continue
  • Kilo Code
  • Junie
  • Trae
  • OpenHands
  • Qwen Code
  • AI Agent
  • Tabnine Chat
  • Sourcegraph Cody
  • Amazon Q
  • Firebase Genkit
  • Sweep
  • Devin
  • Phind
  • Mentat
  • Claude Desktop
  • Aider
  • GPT Pilot
  • Plandex
  • bolt.new
  • v0
  • Replit Agent
  • Anse
  • TypingMind
  • BetterChatGPT
  • LibreChat
  • Warp AI
  • Pieces
  • Fig AI
Embed badge
vskill VERIFIED badge
[![vskill VERIFIED](https://verified-skill.com/api/v1/skills/manutej/luxor-claude-marketplace/asyncio-concurrency-patterns/badge)](https://verified-skill.com/skills/manutej/luxor-claude-marketplace/asyncio-concurrency-patterns)

Something wrong with this skill or its page? Tell us.