randexcpy
is a powerful Python library designed for executing actions at random times within a specified duration. It's perfect for scenarios like load testing, simulating real-world events, or implementing backoff strategies where randomness is key.
- Random Execution: Execute actions at random intervals within a defined time frame.
- Synchronous & Asynchronous: Supports both synchronous and asynchronous task execution.
- Customizable Randomness: Use your own random seed for reproducible results.
- Error Handling: Built-in support for context-based cancellation and error management.
- Lightweight: No external dependencies required, making it easy to integrate into any project.
- Load Testing: Simulate real-world scenarios by introducing randomness in task execution.
- Exponential Backoff: Implement retry strategies with randomized delays to avoid system overloads.
- Chaos Engineering: Introduce controlled randomness to test the resilience of your systems.
- Event Simulation: Model user behavior or system events that occur at unpredictable times.
Install randexcpy
using pip:
pip install randexcpy
Alternatively, you can install the latest development version directly from GitHub:
pip install git+https://github.com/copleftdev/randexcpy.git
from randexcpy import Executor
executor = Executor(max_duration="5s")
# Execute a simple action
result = executor.execute(lambda: "Hello, world!")
print(result) # Output: Hello, world!
from randexcpy import Executor
executor = Executor(max_duration="10s")
# Execute an action asynchronously
async_result = executor.execute_async(lambda: "Hello, async world!")
# Get the result with a timeout
print(async_result.get(timeout=5)) # Output: Hello, async world!
from randexcpy import Executor
# Use a custom random seed for reproducibility
executor = Executor(max_duration="3s", seed=42)
result = executor.execute(lambda: "Reproducible result!")
print(result) # Output will be consistent due to the seed
The Executor
class is the core of randexcpy
, providing methods for executing actions at random intervals.
-
Parameters:
max_duration
(str | timedelta): The maximum duration within which the action must be executed. Example formats: "10s", "5m", "1h".seed
(Optional[int]): A seed for the random number generator (useful for testing and reproducibility).
-
Raises:
ValueError
: Ifmax_duration
is not a valid duration string ortimedelta
.
- Description: Executes the given action synchronously within a random time frame.
- Parameters:
action
(Callable[[], T]): The action to execute.timeout
(Optional[float]): Maximum time to wait for the action to complete.
- Returns: The result of the action.
- Raises:
ExecutionError
: If the action fails or execution exceeds the timeout.
- Description: Executes the given action asynchronously within a random time frame.
- Parameters:
action
(Callable[[], T]): The action to execute.
- Returns: A
Result
object containing the outcome of the execution.
The Result
class represents the outcome of an asynchronous execution.
- Description: Retrieves the result of the asynchronous execution.
- Parameters:
timeout
(Optional[float]): Maximum time to wait for the result.
- Returns: The result of the execution.
- Raises:
ExecutionError
: If the action failed or the result is not available within the timeout.
import requests
from randexcpy import Executor
def make_request():
response = requests.get("https://example.com")
return response.status_code
executor = Executor(max_duration="10s")
for _ in range(10):
status_code = executor.execute(make_request)
print(f"Received status code: {status_code}")
import time
from randexcpy import Executor
def log_event():
print(f"Event logged at {time.time()}")
executor = Executor(max_duration="5s")
# Log events asynchronously
for _ in range(5):
result = executor.execute_async(log_event)
result.get(timeout=10) # Wait for the result with a timeout
This project adheres to strict code quality standards using black
, isort
, flake8
, and pylint
.
To check the code style and linting:
black .
isort .
flake8 .
pylint randexcpy tests
-
Clone the repository:
git clone https://github.com/copleftdev/randexcpy.git cd randexcpy
-
Create a virtual environment:
python -m venv .venv source .venv/bin/activate # On Windows use `.venv\Scripts\activate`
-
Install dependencies:
pip install --upgrade pip pip install -r requirements.txt
-
Run tests:
pytest
We welcome contributions! To get started:
- Fork the repository.
- Create a new branch (
git checkout -b feature/your-feature
). - Commit your changes (
git commit -m 'Add new feature'
). - Push to the branch (
git push origin feature/your-feature
). - Open a pull request.
randexcpy
is licensed under the MIT License. See the LICENSE file for more information.
For any inquiries, feel free to reach out to dj@codetestcode.io.
This project is under active development. New features, bug fixes, and enhancements are regularly added. Stay tuned for updates!