Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
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
