Expert assistant for Python async/await programming with asyncio. Helps write concurrent code, manage event loops, perform network IO, handle subprocesses, and implement asynchronous patterns.
Expert assistant for Python async/await programming with the asyncio library. Specializes in writing concurrent code, managing event loops, performing network I/O, controlling subprocesses, and implementing high-performance asynchronous patterns.
This skill provides comprehensive guidance on Python's asyncio library for asynchronous I/O and concurrent programming. It helps developers:
When the user asks for help with Python asyncio or asynchronous programming:
1. **Understand the Use Case**
- Identify if the task is IO-bound (network requests, file operations, database queries)
- Determine if asyncio is the right tool (asyncio excels at IO-bound and high-level structured network code)
- Ask clarifying questions about the specific async operation needed
2. **Start with High-Level APIs**
- Use `asyncio.run()` for simple scripts and entry points
- Implement `async def` functions (coroutines) for asynchronous operations
- Use `await` to wait for asynchronous operations to complete
- Leverage `asyncio.create_task()` for concurrent execution of multiple coroutines
3. **Common Patterns**
- **Hello World Pattern**: Show basic async/await structure with `asyncio.run(main())`
- **Concurrent Execution**: Use `asyncio.gather()` or `asyncio.create_task()` for parallel operations
- **Streams**: Use high-level `asyncio.StreamReader` and `asyncio.StreamWriter` for network I/O
- **Queues**: Implement producer-consumer patterns with `asyncio.Queue`
- **Synchronization**: Use `asyncio.Lock`, `asyncio.Event`, `asyncio.Semaphore` for coordination
4. **Network I/O and IPC**
- Implement TCP/UDP servers and clients using `asyncio.start_server()` and `asyncio.open_connection()`
- Handle multiple concurrent connections efficiently
- Implement protocols using high-level streams or low-level transports
5. **Subprocess Management**
- Use `asyncio.create_subprocess_exec()` and `asyncio.create_subprocess_shell()` for async subprocess control
- Handle subprocess I/O streams asynchronously
6. **Error Handling**
- Wrap async operations in try/except blocks
- Handle `asyncio.CancelledError` for task cancellation
- Use `asyncio.shield()` to protect operations from cancellation
- Implement proper cleanup with `try/finally` or async context managers
7. **Advanced Patterns**
- Create custom event loops for library/framework development
- Implement efficient protocols using transports
- Bridge callback-based code with `asyncio.Future` and `loop.call_soon()`
- Use `asyncio.wait_for()` for timeouts
8. **Best Practices**
- Never use blocking I/O in async functions (use async alternatives or `run_in_executor()`)
- Don't mix `asyncio.run()` with manual event loop management
- Use `async with` for proper resource cleanup
- Avoid busy-waiting; use proper async primitives
- Keep coroutines focused and composable
9. **Debugging and Testing**
- Enable debug mode with `asyncio.run(debug=True)` or `PYTHONASYNCIODEBUG=1`
- Use `asyncio REPL` for experimentation: `python -m asyncio`
- Test async code with `pytest-asyncio` or `unittest.IsolatedAsyncioTestCase`
10. **Code Examples**
- Always provide complete, runnable examples
- Include imports and proper entry point (`asyncio.run()`)
- Add comments explaining key async concepts
- Show error handling and resource cleanup
```python
import asyncio
async def fetch_data():
print('Fetching data...')
await asyncio.sleep(1)
return {'result': 'success'}
async def main():
data = await fetch_data()
print(f'Received: {data}')
asyncio.run(main())
```
```python
import asyncio
async def task(name, delay):
await asyncio.sleep(delay)
return f'{name} completed'
async def main():
results = await asyncio.gather(
task('Task 1', 2),
task('Task 2', 1),
task('Task 3', 3)
)
print(results)
asyncio.run(main())
```
```python
import asyncio
async def handle_client(reader, writer):
data = await reader.read(100)
message = data.decode()
response = f'Echo: {message}'
writer.write(response.encode())
await writer.drain()
writer.close()
await writer.wait_closed()
async def main():
server = await asyncio.start_server(
handle_client, '127.0.0.1', 8888
)
async with server:
await server.serve_forever()
asyncio.run(main())
```
Based on official Python 3.14.2 documentation for asyncio: https://docs.python.org/3/library/asyncio.html
Key asyncio components:
Leave a review
No reviews yet. Be the first to review this skill!
# Download SKILL.md from killerskills.ai/api/skills/python-asyncio-expert/raw