371 lines
16 KiB
Plaintext
371 lines
16 KiB
Plaintext
---
|
|
description: This rule provides comprehensive guidelines and best practices for utilizing the asyncio library in Python, covering code organization, performance, security, testing, and common pitfalls.
|
|
globs: *.py
|
|
---
|
|
# asyncio Best Practices and Coding Standards
|
|
|
|
This document outlines comprehensive best practices for using the `asyncio` library in Python. It covers various aspects of asyncio development, including code organization, common patterns, performance considerations, security, testing, common pitfalls, and tooling.
|
|
|
|
## Library Information:
|
|
|
|
- Name: asyncio
|
|
- Tags: python, async, standard-library
|
|
|
|
## 1. Code Organization and Structure
|
|
|
|
Effective code organization is crucial for maintainability and scalability when working with asyncio.
|
|
|
|
### 1.1 Directory Structure Best Practices
|
|
|
|
A well-defined directory structure helps in organizing different parts of your asyncio application.
|
|
|
|
|
|
project_root/
|
|
├── src/
|
|
│ ├── __init__.py
|
|
│ ├── main.py # Entry point of the application
|
|
│ ├── modules/
|
|
│ │ ├── __init__.py
|
|
│ │ ├── networking.py # Handles networking tasks
|
|
│ │ ├── processing.py # Data processing tasks
|
|
│ │ └── utils.py # Utility functions
|
|
│ ├── config.py # Configuration settings
|
|
├── tests/
|
|
│ ├── __init__.py
|
|
│ ├── test_networking.py
|
|
│ ├── test_processing.py
|
|
│ └── conftest.py # Fixtures and configuration for pytest
|
|
├── requirements.txt # Project dependencies
|
|
├── pyproject.toml # Project metadata and build system
|
|
├── README.md # Project documentation
|
|
|
|
|
|
### 1.2 File Naming Conventions
|
|
|
|
Consistent file naming conventions enhance readability and maintainability.
|
|
|
|
- Use descriptive names that reflect the module's purpose.
|
|
- Prefer lowercase with underscores (snake_case) for file names (e.g., `async_utils.py`, `data_handler.py`).
|
|
- Test files should follow the `test_*.py` pattern for pytest compatibility.
|
|
|
|
### 1.3 Module Organization
|
|
|
|
Divide your code into logical modules based on functionality.
|
|
|
|
- Group related functions and classes within the same module.
|
|
- Use `__init__.py` files to make directories into Python packages.
|
|
- Employ relative imports for internal modules (`from . import utils`).
|
|
- Use absolute imports for external libraries (`import aiohttp`).
|
|
|
|
python
|
|
# src/modules/networking.py
|
|
import asyncio
|
|
import aiohttp
|
|
|
|
async def fetch_data(url: str) -> str:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url) as response:
|
|
return await response.text()
|
|
|
|
|
|
### 1.4 Component Architecture Recommendations
|
|
|
|
Consider using a layered architecture to separate concerns.
|
|
|
|
- **Presentation Layer**: Handles user interface or API endpoints.
|
|
- **Service Layer**: Contains business logic and orchestrates tasks.
|
|
- **Data Access Layer**: Manages data persistence and retrieval.
|
|
- **Infrastructure Layer**: Provides supporting services like logging and configuration.
|
|
|
|
### 1.5 Code Splitting Strategies
|
|
|
|
Split large modules into smaller, more manageable files.
|
|
|
|
- Use functions and classes to encapsulate specific tasks.
|
|
- Consider splitting modules based on logical boundaries (e.g., one module for database interactions, another for API calls).
|
|
- Refactor complex coroutines into smaller, reusable coroutines.
|
|
|
|
## 2. Common Patterns and Anti-patterns
|
|
|
|
Understanding common patterns and anti-patterns helps in writing efficient and maintainable asyncio code.
|
|
|
|
### 2.1 Design Patterns Specific to asyncio
|
|
|
|
- **Producer-Consumer**: Use `asyncio.Queue` to manage tasks between producers and consumers.
|
|
- **Worker Pool**: Create a pool of worker coroutines to process tasks concurrently.
|
|
- **Pub-Sub**: Implement a publish-subscribe pattern using queues or custom event handling.
|
|
|
|
#### Producer-Consumer Example
|
|
|
|
python
|
|
import asyncio
|
|
|
|
async def producer(queue: asyncio.Queue, data: list):
|
|
for item in data:
|
|
await queue.put(item)
|
|
print(f"Produced: {item}")
|
|
await queue.put(None) # Signal end of production
|
|
|
|
async def consumer(queue: asyncio.Queue):
|
|
while True:
|
|
item = await queue.get()
|
|
if item is None:
|
|
break
|
|
print(f"Consumed: {item}")
|
|
queue.task_done()
|
|
|
|
async def main():
|
|
queue = asyncio.Queue()
|
|
data = [1, 2, 3, 4, 5]
|
|
producer_task = asyncio.create_task(producer(queue, data))
|
|
consumer_task = asyncio.create_task(consumer(queue))
|
|
|
|
await asyncio.gather(producer_task, consumer_task)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|
|
|
|
|
|
### 2.2 Recommended Approaches for Common Tasks
|
|
|
|
- **Making HTTP Requests**: Use `aiohttp` for non-blocking HTTP requests.
|
|
- **Reading/Writing Files**: Use `aiofiles` for asynchronous file I/O.
|
|
- **Database Operations**: Utilize async database drivers like `aiopg` or `asyncpg`.
|
|
- **Task Scheduling**: Use `asyncio.create_task` or `asyncio.gather` to manage concurrent tasks.
|
|
|
|
### 2.3 Anti-patterns and Code Smells to Avoid
|
|
|
|
- **Blocking Calls**: Avoid using blocking functions like `time.sleep` or `requests` in coroutines. Use `asyncio.sleep` and `aiohttp` instead.
|
|
- **Long-Running Coroutines**: Break down long-running coroutines into smaller, awaitable chunks to avoid blocking the event loop.
|
|
- **Ignoring Exceptions**: Always handle exceptions properly to prevent unexpected crashes.
|
|
- **Over-using Locks**: Excessive use of locks can reduce concurrency. Consider using queues or other synchronization primitives.
|
|
- **Unnecessary Context Switching**: Minimize context switches by optimizing code and reducing unnecessary `await` calls.
|
|
|
|
### 2.4 State Management Best Practices
|
|
|
|
- **Immutable Data**: Prefer immutable data structures to avoid race conditions.
|
|
- **Thread-Safe Data Structures**: Use thread-safe data structures from the `queue` or `collections` modules when sharing data between coroutines.
|
|
- **Avoid Global State**: Minimize the use of global state to reduce complexity and potential conflicts.
|
|
|
|
### 2.5 Error Handling Patterns
|
|
|
|
- **Try-Except Blocks**: Use `try-except` blocks to catch and handle exceptions within coroutines.
|
|
- **`asyncio.gather(..., return_exceptions=True)`**: Use `return_exceptions=True` in `asyncio.gather` to prevent one exception from canceling other tasks.
|
|
- **Logging Errors**: Log exceptions with detailed information for debugging purposes.
|
|
- **Graceful Shutdown**: Implement a mechanism to gracefully shut down the application and release resources.
|
|
|
|
python
|
|
import asyncio
|
|
import logging
|
|
|
|
async def my_coroutine(value):
|
|
try:
|
|
if value < 0:
|
|
raise ValueError("Value must be non-negative")
|
|
await asyncio.sleep(1)
|
|
return value * 2
|
|
except ValueError as e:
|
|
logging.error(f"Error processing {value}: {e}")
|
|
return None
|
|
|
|
async def main():
|
|
results = await asyncio.gather(
|
|
my_coroutine(5), my_coroutine(-1), my_coroutine(10), return_exceptions=True
|
|
)
|
|
print(f"Results: {results}")
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.ERROR)
|
|
asyncio.run(main())
|
|
|
|
|
|
## 3. Performance Considerations
|
|
|
|
Optimizing performance is critical for asyncio applications.
|
|
|
|
### 3.1 Optimization Techniques
|
|
|
|
- **Minimize I/O Operations**: Reduce the number of I/O operations by batching requests or caching data.
|
|
- **Use Efficient Data Structures**: Choose appropriate data structures for specific tasks (e.g., dictionaries for fast lookups).
|
|
- **Avoid Unnecessary Copying**: Minimize copying data to reduce memory usage and improve performance.
|
|
- **Profile Your Code**: Use profiling tools to identify performance bottlenecks.
|
|
- **Use `uvloop`**: Consider using `uvloop`, a fast, drop-in replacement for the default asyncio event loop.
|
|
|
|
python
|
|
try:
|
|
import uvloop
|
|
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
|
|
print("Using uvloop")
|
|
except ImportError:
|
|
print("uvloop not installed, using default asyncio loop")
|
|
|
|
|
|
### 3.2 Memory Management
|
|
|
|
- **Resource Management**: Properly release resources (e.g., file handles, database connections) when they are no longer needed.
|
|
- **Use Generators**: Use generators to process large datasets in chunks.
|
|
- **Limit Object Creation**: Reduce the creation of unnecessary objects to minimize memory overhead.
|
|
|
|
### 3.3 Lazy Loading Strategies
|
|
|
|
- **Import on Demand**: Import modules only when they are needed to reduce startup time.
|
|
- **Load Data Lazily**: Load data only when it is accessed to reduce initial memory usage.
|
|
|
|
## 4. Security Best Practices
|
|
|
|
Securing asyncio applications is essential for protecting against vulnerabilities.
|
|
|
|
### 4.1 Common Vulnerabilities and Prevention
|
|
|
|
- **Injection Attacks**: Sanitize user inputs to prevent SQL injection, command injection, and other injection attacks.
|
|
- **Cross-Site Scripting (XSS)**: Encode user-generated content to prevent XSS attacks.
|
|
- **Denial of Service (DoS)**: Implement rate limiting and input validation to prevent DoS attacks.
|
|
- **Man-in-the-Middle (MitM) Attacks**: Use TLS/SSL for secure communication to prevent MitM attacks.
|
|
|
|
### 4.2 Input Validation
|
|
|
|
- **Validate All Inputs**: Validate all user inputs and data received from external sources.
|
|
- **Use Regular Expressions**: Use regular expressions to validate input formats.
|
|
- **Limit Input Length**: Restrict the length of input fields to prevent buffer overflows.
|
|
|
|
### 4.3 Authentication and Authorization
|
|
|
|
- **Use Strong Authentication**: Implement strong authentication mechanisms (e.g., multi-factor authentication).
|
|
- **Implement Authorization**: Implement authorization checks to ensure users only have access to authorized resources.
|
|
- **Store Passwords Securely**: Hash passwords using strong hashing algorithms (e.g., bcrypt or Argon2).
|
|
- **Use JWTs**: Employ JSON Web Tokens (JWTs) for stateless authentication.
|
|
|
|
### 4.4 Data Protection Strategies
|
|
|
|
- **Encrypt Sensitive Data**: Encrypt sensitive data at rest and in transit.
|
|
- **Use Secure Protocols**: Use secure protocols (e.g., HTTPS, SSH) for communication.
|
|
- **Regularly Audit Security**: Conduct regular security audits to identify and address vulnerabilities.
|
|
|
|
### 4.5 Secure API Communication
|
|
|
|
- **Use HTTPS**: Always use HTTPS for API communication.
|
|
- **Validate API Responses**: Validate API responses to ensure data integrity.
|
|
- **Implement Rate Limiting**: Implement rate limiting to prevent abuse.
|
|
|
|
## 5. Testing Approaches
|
|
|
|
Testing is crucial for ensuring the reliability of asyncio applications.
|
|
|
|
### 5.1 Unit Testing Strategies
|
|
|
|
- **Test Coroutines in Isolation**: Use `asyncio.run` or `pytest-asyncio` to test coroutines in isolation.
|
|
- **Mock External Dependencies**: Use mocking libraries like `unittest.mock` to mock external dependencies.
|
|
- **Assert Expected Outcomes**: Use assertions to verify expected outcomes and error conditions.
|
|
|
|
### 5.2 Integration Testing
|
|
|
|
- **Test Interactions Between Components**: Test interactions between different components of the application.
|
|
- **Use Real Dependencies**: Use real dependencies (e.g., databases, APIs) in a controlled environment.
|
|
|
|
### 5.3 End-to-End Testing
|
|
|
|
- **Simulate Real User Scenarios**: Simulate real user scenarios to test the entire application flow.
|
|
- **Use Test Automation Frameworks**: Use test automation frameworks like Selenium or Playwright.
|
|
|
|
### 5.4 Test Organization
|
|
|
|
- **Organize Tests by Module**: Organize tests into separate files that correspond to the application modules.
|
|
- **Use Descriptive Test Names**: Use descriptive test names that clearly indicate what is being tested.
|
|
|
|
### 5.5 Mocking and Stubbing Techniques
|
|
|
|
- **Mock Asynchronous Functions**: Use `asyncio.iscoroutinefunction` to check if a function is a coroutine before mocking it.
|
|
- **Patch External Dependencies**: Use `unittest.mock.patch` to replace external dependencies with mock objects.
|
|
- **Use Asynchronous Mocks**: Use asynchronous mocks to simulate asynchronous behavior.
|
|
|
|
python
|
|
import asyncio
|
|
import unittest.mock
|
|
import pytest
|
|
|
|
async def fetch_data(url: str) -> str:
|
|
# This is just a placeholder; in real code, it would use aiohttp
|
|
await asyncio.sleep(0.1) # Simulate I/O delay
|
|
return f"Data from {url}"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_data():
|
|
with unittest.mock.patch("__main__.fetch_data") as mock_fetch_data:
|
|
mock_fetch_data.return_value = "Mocked data"
|
|
result = await fetch_data("http://example.com")
|
|
assert result == "Mocked data"
|
|
mock_fetch_data.assert_called_once_with("http://example.com")
|
|
|
|
|
|
## 6. Common Pitfalls and Gotchas
|
|
|
|
Being aware of common pitfalls helps in avoiding mistakes when using asyncio.
|
|
|
|
### 6.1 Frequent Mistakes
|
|
|
|
- **Mixing Synchronous and Asynchronous Code**: Avoid mixing synchronous and asynchronous code in the same coroutine.
|
|
- **Forgetting to Await**: Ensure that you `await` all awaitable objects.
|
|
- **Blocking the Event Loop**: Avoid blocking the event loop with long-running synchronous operations.
|
|
- **Ignoring Task Cancellation**: Handle task cancellation properly to prevent resource leaks.
|
|
- **Not Handling Exceptions**: Always handle exceptions properly to prevent unexpected crashes.
|
|
|
|
### 6.2 Edge Cases
|
|
|
|
- **Task Timeouts**: Implement task timeouts to prevent tasks from running indefinitely.
|
|
- **Resource Limits**: Set resource limits (e.g., maximum number of connections) to prevent resource exhaustion.
|
|
- **Signal Handling**: Handle signals (e.g., SIGINT, SIGTERM) to gracefully shut down the application.
|
|
|
|
### 6.3 Version-Specific Issues
|
|
|
|
- **asyncio API Changes**: Be aware of API changes between different versions of asyncio.
|
|
- **Python 3.7+**: Use Python 3.7 or later to take advantage of the latest asyncio features (e.g., `asyncio.run`).
|
|
|
|
### 6.4 Compatibility Concerns
|
|
|
|
- **Third-Party Libraries**: Ensure that third-party libraries are compatible with asyncio.
|
|
- **Event Loop Implementations**: Be aware of compatibility issues between different event loop implementations (e.g., `uvloop`).
|
|
|
|
### 6.5 Debugging Strategies
|
|
|
|
- **Enable Debug Mode**: Enable asyncio debug mode to get more detailed information about tasks and coroutines.
|
|
- **Use Logging**: Use logging to track the execution flow and identify potential issues.
|
|
- **Use Debuggers**: Use debuggers like `pdb` or IDE-integrated debuggers to step through code and inspect variables.
|
|
- **Inspect Task State**: Inspect the state of tasks using `asyncio.Task.all_tasks()` to identify stuck or failing tasks.
|
|
|
|
## 7. Tooling and Environment
|
|
|
|
Using the right tools and environment can greatly improve the development experience.
|
|
|
|
### 7.1 Recommended Development Tools
|
|
|
|
- **IDE**: Use an IDE like VS Code, PyCharm, or Sublime Text with Python support.
|
|
- **Linters**: Use linters like `flake8` or `pylint` to enforce coding standards.
|
|
- **Formatters**: Use formatters like `black` or `autopep8` to automatically format code.
|
|
- **Debuggers**: Use debuggers like `pdb` or IDE-integrated debuggers to step through code and inspect variables.
|
|
|
|
### 7.2 Build Configuration
|
|
|
|
- **Use `pyproject.toml`**: Configure the project using a `pyproject.toml` file to specify build dependencies and settings.
|
|
- **Specify Dependencies**: Specify project dependencies in `requirements.txt` or `pyproject.toml`.
|
|
- **Use Virtual Environments**: Use virtual environments to isolate project dependencies.
|
|
|
|
### 7.3 Linting and Formatting
|
|
|
|
- **Configure Linters**: Configure linters to enforce coding standards (e.g., PEP 8).
|
|
- **Configure Formatters**: Configure formatters to automatically format code.
|
|
- **Use Pre-commit Hooks**: Use pre-commit hooks to automatically run linters and formatters before committing code.
|
|
|
|
### 7.4 Deployment Best Practices
|
|
|
|
- **Use a Production-Ready Event Loop**: Use a production-ready event loop like `uvloop`.
|
|
- **Use a Process Manager**: Use a process manager like systemd or Supervisor to manage the application process.
|
|
- **Use a Load Balancer**: Use a load balancer to distribute traffic across multiple instances of the application.
|
|
- **Monitor Application Health**: Monitor the application's health and performance using metrics and alerts.
|
|
|
|
### 7.5 CI/CD Integration
|
|
|
|
- **Automate Tests**: Automate unit, integration, and end-to-end tests in the CI/CD pipeline.
|
|
- **Automate Linting and Formatting**: Automate linting and formatting in the CI/CD pipeline.
|
|
- **Automate Deployment**: Automate deployment to production environments.
|
|
|
|
By adhering to these best practices, you can build robust, efficient, and maintainable asyncio applications in Python. |