Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Python's asyncio library enables writing concurrent code using async/await syntax. It's ideal for I/O-bound operations like HTTP requests, database queries, file operations, and WebSocket connections. asyncio provides non-blocking execution without the complexity of threading or multiprocessing.
Key Features:
async/await syntax for readable concurrent code
Event loop for managing concurrent operations
Tasks for running multiple coroutines concurrently
import asyncio
asyncdefbackground_task(name):
"""Long-running background task."""for i inrange(5):
print(f"{name}: iteration {i}")
await asyncio.sleep(1)
returnf"{name} complete"asyncdefmain():
# Create task (starts immediately)
task1 = asyncio.create_task(background_task("Task-1"))
task2 = asyncio.create_task(background_task("Task-2"))
# Do other work while tasks runprint("Main: doing other work")
await asyncio.sleep(2)
# Wait for tasks to complete
result1 = await task1
result2 = await task2
print(f"Results: {result1}, {result2}")
asyncio.run(main())
4. Error Handling in Async Code
import asyncio
asyncdefrisky_operation(fail=False):
"""Operation that might fail."""await asyncio.sleep(1)
if fail:
raise ValueError("Operation failed")
return"Success"asyncdefhandle_errors():
# Individual try/excepttry:
result = await risky_operation(fail=True)
except ValueError as e:
print(f"Caught error: {e}")
result = "Fallback value"# Gather with error handling
results = await asyncio.gather(
risky_operation(fail=False),
risky_operation(fail=True),
risky_operation(fail=False),
return_exceptions=True# Return exceptions instead of raising
)
for i, result inenumerate(results):
ifisinstance(result, Exception):
print(f"Task {i} failed: {result}")
else:
print(f"Task {i} succeeded: {result}")
asyncio.run(handle_errors())
Event Loop Fundamentals
1. Event Loop Lifecycle
import asyncio
# Modern approach (Python 3.7+)asyncdefmain():
print("Main coroutine")
await asyncio.sleep(1)
asyncio.run(main()) # Creates loop, runs main, closes loop# Manual loop management (advanced use cases)asyncdefmanual_example():
loop = asyncio.get_event_loop()
# Schedule coroutine
task = loop.create_task(some_coroutine())
# Schedule callback
loop.call_later(5, callback_function)
# Run until complete
result = await task
return result
# Get current event loopasyncdefget_current_loop():
loop = asyncio.get_running_loop()
print(f"Loop: {loop}")
# Schedule callback in event loop
loop.call_soon(lambda: print("Callback executed"))
await asyncio.sleep(0) # Let callback execute
2. Loop Scheduling and Callbacks
import asyncio
from datetime import datetime
defcallback(name, loop):
"""Callback function (not async)."""print(f"{datetime.now()}: {name} callback executed")
# Stop loop after callback# loop.stop()asyncdefschedule_callbacks():
loop = asyncio.get_running_loop()
# Schedule immediate callback
loop.call_soon(callback, "Immediate", loop)
# Schedule callback after delay
loop.call_later(2, callback, "Delayed 2s", loop)
# Schedule callback at specific time
loop.call_at(loop.time() + 3, callback, "Delayed 3s", loop)
# Wait for callbacks to executeawait asyncio.sleep(5)
asyncio.run(schedule_callbacks())
3. Running Blocking Code
import asyncio
import time
defblocking_io():
"""CPU-intensive or blocking I/O operation."""print("Blocking operation started")
time.sleep(2) # Blocks threadprint("Blocking operation complete")
return"Blocking result"asyncdefrun_in_executor():
"""Run blocking code in thread pool."""
loop = asyncio.get_running_loop()
# Run in default executor (thread pool)
result = await loop.run_in_executor(
None, # Use default executor
blocking_io
)
print(f"Result: {result}")
# Run blocking operations concurrentlyasyncdefconcurrent_blocking():
loop = asyncio.get_running_loop()
# These run in thread pool, don't block event loop
results = await asyncio.gather(
loop.run_in_executor(None, blocking_io),
loop.run_in_executor(None, blocking_io),
loop.run_in_executor(None, blocking_io)
)
print(f"All results: {results}")
asyncio.run(concurrent_blocking())
Asyncio Primitives
1. Locks for Mutual Exclusion
import asyncio
# Shared resource
counter = 0
lock = asyncio.Lock()
asyncdefincrement_with_lock(name):
"""Increment counter with lock protection."""global counter
asyncwith lock:
# Critical section - only one task at a timeprint(f"{name}: acquired lock")
current = counter
await asyncio.sleep(0.1) # Simulate processing
counter = current + 1print(f"{name}: released lock, counter={counter}")
asyncdefincrement_without_lock(name):
"""Increment without lock - race condition!"""global counter
current = counter
await asyncio.sleep(0.1) # Race condition window
counter = current + 1print(f"{name}: counter={counter}")
asyncdeftest_locks():
global counter
# Without lock (race condition)
counter = 0await asyncio.gather(
increment_without_lock("Task-1"),
increment_without_lock("Task-2"),
increment_without_lock("Task-3")
)
print(f"Without lock: {counter}") # Often wrong (< 3)# With lock (correct)
counter = 0await asyncio.gather(
increment_with_lock("Task-1"),
increment_with_lock("Task-2"),
increment_with_lock("Task-3")
)
print(f"With lock: {counter}") # Always 3
asyncio.run(test_locks())
2. Semaphores for Resource Limiting
import asyncio
# Limit concurrent operations
semaphore = asyncio.Semaphore(2) # Max 2 concurrentasyncdeflimited_operation(name):
"""Operation limited by semaphore."""print(f"{name}: waiting for semaphore")
asyncwith semaphore:
print(f"{name}: acquired semaphore")
await asyncio.sleep(2) # Simulate workprint(f"{name}: releasing semaphore")
asyncdeftest_semaphore():
# Create 5 tasks, but only 2 run concurrentlyawait asyncio.gather(
limited_operation("Task-1"),
limited_operation("Task-2"),
limited_operation("Task-3"),
limited_operation("Task-4"),
limited_operation("Task-5")
)
asyncio.run(test_semaphore())
# Only 2 tasks hold semaphore at any time
3. Events for Signaling
import asyncio
event = asyncio.Event()
asyncdefwaiter(name):
"""Wait for event to be set."""print(f"{name}: waiting for event")
await event.wait() # Block until event is setprint(f"{name}: event received!")
asyncdefsetter():
"""Set event after delay."""await asyncio.sleep(2)
print("Setter: setting event")
event.set() # Wake up all waitersasyncdeftest_event():
# Create waitersawait asyncio.gather(
waiter("Waiter-1"),
waiter("Waiter-2"),
waiter("Waiter-3"),
setter()
)
asyncio.run(test_event())
4. Queues for Task Distribution
import asyncio
import random
asyncdefproducer(queue, name):
"""Produce items and add to queue."""for i inrange(5):
item = f"{name}-item-{i}"await queue.put(item)
print(f"{name}: produced {item}")
await asyncio.sleep(random.uniform(0.1, 0.5))
# Signal completionawait queue.put(None)
asyncdefconsumer(queue, name):
"""Consume items from queue."""whileTrue:
item = await queue.get() # Block until item availableif item isNone: # Shutdown signal
queue.task_done()
breakprint(f"{name}: consumed {item}")
await asyncio.sleep(random.uniform(0.2, 0.8))
queue.task_done()
asyncdeftest_queue():
queue = asyncio.Queue(maxsize=10)
# Create producers and consumersawait asyncio.gather(
producer(queue, "Producer-1"),
producer(queue, "Producer-2"),
consumer(queue, "Consumer-1"),
consumer(queue, "Consumer-2"),
consumer(queue, "Consumer-3")
)
# Wait for all items to be processedawait queue.join()
print("All tasks complete")
asyncio.run(test_queue())
5. Condition Variables
import asyncio
condition = asyncio.Condition()
items = []
asyncdefconsumer(name):
"""Wait for items to be available."""asyncwith condition:
# Wait until items are availableawait condition.wait_for(lambda: len(items) > 0)
item = items.pop(0)
print(f"{name}: consumed {item}")
asyncdefproducer(name):
"""Add items and notify consumers."""asyncwith condition:
item = f"{name}-item"
items.append(item)
print(f"{name}: produced {item}")
# Notify one waiting consumer
condition.notify(n=1)
# Or notify all: condition.notify_all()asyncdeftest_condition():
await asyncio.gather(
consumer("Consumer-1"),
consumer("Consumer-2"),
producer("Producer-1"),
producer("Producer-2")
)
asyncio.run(test_condition())
Async HTTP with aiohttp
1. Basic HTTP Client
import asyncio
import aiohttp
asyncdeffetch_url(session, url):
"""Fetch single URL."""asyncwith session.get(url) as response:
status = response.status
text = await response.text()
return {"url": url, "status": status, "length": len(text)}
asyncdeffetch_multiple_urls():
"""Fetch multiple URLs concurrently."""
urls = [
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/2",
"https://httpbin.org/json",
]
asyncwith aiohttp.ClientSession() as session:
# Concurrent requests
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(f"{result['url']}: {result['status']} ({result['length']} bytes)")
asyncio.run(fetch_multiple_urls())
2. HTTP Client with Error Handling
import asyncio
import aiohttp
from typing importDict, Anyasyncdeffetch_with_retry(
session: aiohttp.ClientSession,
url: str,
max_retries: int = 3) -> Dict[str, Any]:
"""Fetch URL with retry logic."""for attempt inrange(max_retries):
try:
asyncwith session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
response.raise_for_status() # Raise for 4xx/5xx
data = await response.json()
return {"success": True, "data": data}
except aiohttp.ClientError as e:
print(f"Attempt {attempt + 1} failed: {e}")
if attempt == max_retries - 1:
return {"success": False, "error": str(e)}
# Exponential backoffawait asyncio.sleep(2 ** attempt)
except asyncio.TimeoutError:
print(f"Attempt {attempt + 1} timed out")
if attempt == max_retries - 1:
return {"success": False, "error": "Timeout"}
await asyncio.sleep(2 ** attempt)
asyncdefparallel_api_calls():
"""Make parallel API calls with error handling."""
urls = [
"https://httpbin.org/json",
"https://httpbin.org/status/500", # Will fail"https://httpbin.org/delay/1",
]
asyncwith aiohttp.ClientSession() as session:
results = await asyncio.gather(
*[fetch_with_retry(session, url) for url in urls],
return_exceptions=True# Don't stop on errors
)
for url, result inzip(urls, results):
ifisinstance(result, Exception):
print(f"{url}: Exception - {result}")
elif result["success"]:
print(f"{url}: Success")
else:
print(f"{url}: Failed - {result['error']}")
asyncio.run(parallel_api_calls())
3. HTTP Server with aiohttp
from aiohttp import web
import asyncio
asyncdefhandle_hello(request):
"""Simple GET handler."""
name = request.query.get("name", "World")
return web.json_response({"message": f"Hello, {name}!"})
asyncdefhandle_post(request):
"""POST handler with JSON body."""
data = await request.json()
# Simulate async processingawait asyncio.sleep(1)
return web.json_response({
"received": data,
"status": "processed"
})
asyncdefhandle_stream(request):
"""Streaming response."""
response = web.StreamResponse()
await response.prepare(request)
for i inrange(10):
await response.write(f"Chunk {i}\n".encode())
await asyncio.sleep(0.5)
await response.write_eof()
return response
# Create application
app = web.Application()
app.router.add_get("/hello", handle_hello)
app.router.add_post("/process", handle_post)
app.router.add_get("/stream", handle_stream)
# Run serverif __name__ == "__main__":
web.run_app(app, host="0.0.0.0", port=8080)
4. WebSocket Client
import asyncio
import aiohttp
asyncdefwebsocket_client():
"""Connect to WebSocket server."""
url = "wss://echo.websocket.org"asyncwith aiohttp.ClientSession() as session:
asyncwith session.ws_connect(url) as ws:
# Send messagesawait ws.send_str("Hello WebSocket")
await ws.send_json({"type": "greeting", "data": "test"})
# Receive messagesasyncfor msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
print(f"Received: {msg.data}")
if msg.data == "close":
await ws.close()
breakelif msg.type == aiohttp.WSMsgType.ERROR:
print(f"Error: {ws.exception()}")
break
asyncio.run(websocket_client())
Async Database Operations
1. PostgreSQL with asyncpg
import asyncio
import asyncpg
asyncdefdatabase_operations():
"""Async PostgreSQL operations."""# Create connection pool
pool = await asyncpg.create_pool(
host="localhost",
database="mydb",
user="user",
password="password",
min_size=5,
max_size=20
)
try:
# Acquire connection from poolasyncwith pool.acquire() as conn:
# Execute query
rows = await conn.fetch(
"SELECT id, name, email FROM users WHERE active = $1",
True
)
for row in rows:
print(f"User: {row['name']} ({row['email']})")
# Insert dataawait conn.execute(
"INSERT INTO users (name, email) VALUES ($1, $2)",
"Alice", "alice@example.com"
)
# Transactionasyncwith conn.transaction():
await conn.execute("UPDATE users SET active = $1 WHERE id = $2", False, 1)
await conn.execute("INSERT INTO audit_log (action) VALUES ($1)", "deactivate")
finally:
await pool.close()
asyncio.run(database_operations())
import asyncio
asyncdefcancellable_task():
"""Task that can be cancelled."""try:
for i inrange(10):
print(f"Working: {i}")
await asyncio.sleep(1)
return"Complete"except asyncio.CancelledError:
print("Task was cancelled")
# Cleanupraise# Re-raise to propagate cancellationasyncdefcancel_example():
"""Example of task cancellation."""
task = asyncio.create_task(cancellable_task())
# Let it run for a bitawait asyncio.sleep(3)
# Cancel the task
task.cancel()
try:
await task
except asyncio.CancelledError:
print("Confirmed: task was cancelled")
asyncio.run(cancel_example())
import asyncio
import time
asyncdefproblematic_code():
"""Code with blocking operation."""print("Starting")
# BAD: Blocking sleep
time.sleep(2) # This blocks the event loop!print("Complete")
# Run with debug mode to detect blocking
asyncio.run(problematic_code(), debug=True)
# Warning: Executing <Task> took 2.001 seconds
3. Track Pending Tasks
import asyncio
asyncdeftrack_tasks():
"""Track all pending tasks."""# Get all tasks
tasks = asyncio.all_tasks()
print(f"Total tasks: {len(tasks)}")
for task in tasks:
print(f" - {task.get_name()}: {task}")
# Check if task is doneif task.done():
try:
result = task.result()
print(f" Result: {result}")
except Exception as e:
print(f" Exception: {e}")
# Create some tasksasyncdefmain():
task1 = asyncio.create_task(asyncio.sleep(5), name="sleep-task")
task2 = asyncio.create_task(track_tasks(), name="tracking")
await task2
task1.cancel()
asyncio.run(main())
Testing Async Code
1. pytest-asyncio Setup
# test_async.pyimport pytest
import asyncio
# Mark test as async@pytest.mark.asyncioasyncdeftest_async_function():
"""Test async function."""
result = await some_async_function()
assert result == "expected"@pytest.mark.asyncioasyncdeftest_async_http():
"""Test async HTTP client."""asyncwith aiohttp.ClientSession() as session:
asyncwith session.get("https://httpbin.org/json") as response:
assert response.status == 200
data = await response.json()
assert"slideshow"in data
# Async fixture@pytest.fixtureasyncdefasync_client():
"""Async test fixture."""
client = await create_async_client()
yield client
await client.close()
@pytest.mark.asyncioasyncdeftest_with_fixture(async_client):
"""Test using async fixture."""
result = await async_client.fetch_data()
assert result isnotNone
2. Mocking Async Functions
import pytest
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncioasyncdeftest_with_mock():
"""Test with async mock."""# Create async mock
mock_func = AsyncMock(return_value="mocked result")
result = await mock_func()
assert result == "mocked result"
mock_func.assert_called_once()
@pytest.mark.asyncio@patch("module.async_function", new_callable=AsyncMock)asyncdeftest_with_patch(mock_async):
"""Test with patched async function."""
mock_async.return_value = {"status": "success"}
result = await some_function_that_calls_async()
assert result["status"] == "success"
mock_async.assert_called_once()
Performance Optimization
1. Use asyncio.gather() for Parallelism
import asyncio
import time
asyncdefslow_task(n):
await asyncio.sleep(1)
return n * 2asyncdefoptimized():
"""Parallel execution."""
start = time.time()
# Sequential (slow) - 5 seconds# results = []# for i in range(5):# result = await slow_task(i)# results.append(result)# Parallel (fast) - 1 second
results = await asyncio.gather(*[slow_task(i) for i inrange(5)])
elapsed = time.time() - start
print(f"Time: {elapsed:.2f}s, Results: {results}")
asyncio.run(optimized())
2. Connection Pooling
import asyncio
import aiohttp
# BAD: Create new session for each requestasyncdefbad_pattern():
for i inrange(10):
asyncwith aiohttp.ClientSession() as session:
asyncwith session.get("https://httpbin.org/json") as response:
await response.json()
# GOOD: Reuse session with connection poolasyncdefgood_pattern():
asyncwith aiohttp.ClientSession() as session:
tasks = [
session.get("https://httpbin.org/json")
for i inrange(10)
]
responses = await asyncio.gather(*tasks)
for response in responses:
await response.json()
3. Avoid Blocking Operations
import asyncio
# BAD: Blocking I/O in async functionasyncdefbad_file_read():
withopen("large_file.txt") as f: # Blocks event loop!
data = f.read()
return data
# GOOD: Use async file I/O or run in executorasyncdefgood_file_read():
loop = asyncio.get_running_loop()
# Run blocking operation in thread pool
data = await loop.run_in_executor(
None,
lambda: open("large_file.txt").read()
)
return data
# BETTER: Use aiofiles for async file I/Oimport aiofiles
asyncdefbetter_file_read():
asyncwith aiofiles.open("large_file.txt") as f:
data = await f.read()
return data
