# 🌟 `randexcpy` - Random Execution Library for Python
[](https://badge.fury.io/py/randexcpy)
[](https://github.com/copleftdev/randexcpy/actions)
[](https://opensource.org/licenses/MIT)
`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.
## 🚀 Features
- **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.
## 🎯 Use Cases
- **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.
## 📦 Installation
Install `randexcpy` using pip:
```bash
pip install randexcpy
```
Alternatively, you can install the latest development version directly from GitHub:
```bash
pip install git+https://github.com/copleftdev/randexcpy.git
```
## 🛠️ Usage
### Synchronous Execution
```python
from randexcpy import Executor
executor = Executor(max_duration="5s")
# Execute a simple action
result = executor.execute(lambda: "Hello, world!")
print(result) # Output: Hello, world!
```
### Asynchronous Execution
```python
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!
```
### Customizing Execution with a Random Seed
```python
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
```
## 📖 API Documentation
### `Executor`
The `Executor` class is the core of `randexcpy`, providing methods for executing actions at random intervals.
#### `Executor.__init__(max_duration: Union[str, timedelta], seed: Optional[int] = None)`
- **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`: If `max_duration` is not a valid duration string or `timedelta`.
#### `Executor.execute(action: Callable[[], T], timeout: Optional[float] = None) -> T`
- **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.
#### `Executor.execute_async(action: Callable[[], T]) -> Result`
- **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.
### `Result`
The `Result` class represents the outcome of an asynchronous execution.
#### `Result.get(timeout: Optional[float] = None) -> Any`
- **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.
## 🧪 Examples
### Example: Load Testing with Random Delays
```python
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}")
```
### Example: Asynchronous Event Simulation
```python
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
```
## 🎨 Code Style and Linting
This project adheres to strict code quality standards using `black`, `isort`, `flake8`, and `pylint`.
To check the code style and linting:
```bash
black .
isort .
flake8 .
pylint randexcpy tests
```
## 🔧 Development Setup
1. **Clone the repository**:
```bash
git clone https://github.com/copleftdev/randexcpy.git
cd randexcpy
```
2. **Create a virtual environment**:
```bash
python -m venv .venv
source .venv/bin/activate # On Windows use `.venv\Scripts\activate`
```
3. **Install dependencies**:
```bash
pip install --upgrade pip
pip install -r requirements.txt
```
4. **Run tests**:
```bash
pytest
```
## 🌟 Contributing
We welcome contributions! To get started:
1. Fork the repository.
2. Create a new branch (`git checkout -b feature/your-feature`).
3. Commit your changes (`git commit -m 'Add new feature'`).
4. Push to the branch (`git push origin feature/your-feature`).
5. Open a pull request.
## 📄 License
`randexcpy` is licensed under the MIT License. See the [LICENSE](LICENSE) file for more information.
## 📧 Contact
For any inquiries, feel free to reach out to [dj@codetestcode.io](mailto:dj@codetestcode.io).
---
### 📈 Project Status
This project is under active development. New features, bug fixes, and enhancements are regularly added. Stay tuned for updates!
Raw data
{
"_id": null,
"home_page": "https://github.com/copleftdev/randexcpy",
"name": "randexcpy",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": null,
"author": "Don Johnson",
"author_email": "dj@codetestcode.io",
"download_url": "https://files.pythonhosted.org/packages/70/6b/8c041e03ac0f1a5515cfe2a787693867739467be1aeeeeb663d1af57531b/randexcpy-0.1.0.tar.gz",
"platform": null,
"description": "# \ud83c\udf1f `randexcpy` - Random Execution Library for Python\n\n[](https://badge.fury.io/py/randexcpy)\n[](https://github.com/copleftdev/randexcpy/actions)\n[](https://opensource.org/licenses/MIT)\n\n`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.\n\n## \ud83d\ude80 Features\n\n- **Random Execution**: Execute actions at random intervals within a defined time frame.\n- **Synchronous & Asynchronous**: Supports both synchronous and asynchronous task execution.\n- **Customizable Randomness**: Use your own random seed for reproducible results.\n- **Error Handling**: Built-in support for context-based cancellation and error management.\n- **Lightweight**: No external dependencies required, making it easy to integrate into any project.\n\n## \ud83c\udfaf Use Cases\n\n- **Load Testing**: Simulate real-world scenarios by introducing randomness in task execution.\n- **Exponential Backoff**: Implement retry strategies with randomized delays to avoid system overloads.\n- **Chaos Engineering**: Introduce controlled randomness to test the resilience of your systems.\n- **Event Simulation**: Model user behavior or system events that occur at unpredictable times.\n\n## \ud83d\udce6 Installation\n\nInstall `randexcpy` using pip:\n\n```bash\npip install randexcpy\n```\n\nAlternatively, you can install the latest development version directly from GitHub:\n\n```bash\npip install git+https://github.com/copleftdev/randexcpy.git\n```\n\n## \ud83d\udee0\ufe0f Usage\n\n### Synchronous Execution\n\n```python\nfrom randexcpy import Executor\n\nexecutor = Executor(max_duration=\"5s\")\n\n# Execute a simple action\nresult = executor.execute(lambda: \"Hello, world!\")\nprint(result) # Output: Hello, world!\n```\n\n### Asynchronous Execution\n\n```python\nfrom randexcpy import Executor\n\nexecutor = Executor(max_duration=\"10s\")\n\n# Execute an action asynchronously\nasync_result = executor.execute_async(lambda: \"Hello, async world!\")\n\n# Get the result with a timeout\nprint(async_result.get(timeout=5)) # Output: Hello, async world!\n```\n\n### Customizing Execution with a Random Seed\n\n```python\nfrom randexcpy import Executor\n\n# Use a custom random seed for reproducibility\nexecutor = Executor(max_duration=\"3s\", seed=42)\nresult = executor.execute(lambda: \"Reproducible result!\")\nprint(result) # Output will be consistent due to the seed\n```\n\n## \ud83d\udcd6 API Documentation\n\n### `Executor`\n\nThe `Executor` class is the core of `randexcpy`, providing methods for executing actions at random intervals.\n\n#### `Executor.__init__(max_duration: Union[str, timedelta], seed: Optional[int] = None)`\n\n- **Parameters**:\n - `max_duration` (str | timedelta): The maximum duration within which the action must be executed. Example formats: \"10s\", \"5m\", \"1h\".\n - `seed` (Optional[int]): A seed for the random number generator (useful for testing and reproducibility).\n\n- **Raises**:\n - `ValueError`: If `max_duration` is not a valid duration string or `timedelta`.\n\n#### `Executor.execute(action: Callable[[], T], timeout: Optional[float] = None) -> T`\n\n- **Description**: Executes the given action synchronously within a random time frame.\n- **Parameters**:\n - `action` (Callable[[], T]): The action to execute.\n - `timeout` (Optional[float]): Maximum time to wait for the action to complete.\n- **Returns**: The result of the action.\n- **Raises**:\n - `ExecutionError`: If the action fails or execution exceeds the timeout.\n\n#### `Executor.execute_async(action: Callable[[], T]) -> Result`\n\n- **Description**: Executes the given action asynchronously within a random time frame.\n- **Parameters**:\n - `action` (Callable[[], T]): The action to execute.\n- **Returns**: A `Result` object containing the outcome of the execution.\n\n### `Result`\n\nThe `Result` class represents the outcome of an asynchronous execution.\n\n#### `Result.get(timeout: Optional[float] = None) -> Any`\n\n- **Description**: Retrieves the result of the asynchronous execution.\n- **Parameters**:\n - `timeout` (Optional[float]): Maximum time to wait for the result.\n- **Returns**: The result of the execution.\n- **Raises**:\n - `ExecutionError`: If the action failed or the result is not available within the timeout.\n\n## \ud83e\uddea Examples\n\n### Example: Load Testing with Random Delays\n\n```python\nimport requests\nfrom randexcpy import Executor\n\ndef make_request():\n response = requests.get(\"https://example.com\")\n return response.status_code\n\nexecutor = Executor(max_duration=\"10s\")\n\nfor _ in range(10):\n status_code = executor.execute(make_request)\n print(f\"Received status code: {status_code}\")\n```\n\n### Example: Asynchronous Event Simulation\n\n```python\nimport time\nfrom randexcpy import Executor\n\ndef log_event():\n print(f\"Event logged at {time.time()}\")\n\nexecutor = Executor(max_duration=\"5s\")\n\n# Log events asynchronously\nfor _ in range(5):\n result = executor.execute_async(log_event)\n result.get(timeout=10) # Wait for the result with a timeout\n```\n\n## \ud83c\udfa8 Code Style and Linting\n\nThis project adheres to strict code quality standards using `black`, `isort`, `flake8`, and `pylint`.\n\nTo check the code style and linting:\n\n```bash\nblack .\nisort .\nflake8 .\npylint randexcpy tests\n```\n\n## \ud83d\udd27 Development Setup\n\n1. **Clone the repository**:\n\n ```bash\n git clone https://github.com/copleftdev/randexcpy.git\n cd randexcpy\n ```\n\n2. **Create a virtual environment**:\n\n ```bash\n python -m venv .venv\n source .venv/bin/activate # On Windows use `.venv\\Scripts\\activate`\n ```\n\n3. **Install dependencies**:\n\n ```bash\n pip install --upgrade pip\n pip install -r requirements.txt\n ```\n\n4. **Run tests**:\n\n ```bash\n pytest\n ```\n\n## \ud83c\udf1f Contributing\n\nWe welcome contributions! To get started:\n\n1. Fork the repository.\n2. Create a new branch (`git checkout -b feature/your-feature`).\n3. Commit your changes (`git commit -m 'Add new feature'`).\n4. Push to the branch (`git push origin feature/your-feature`).\n5. Open a pull request.\n\n## \ud83d\udcc4 License\n\n`randexcpy` is licensed under the MIT License. See the [LICENSE](LICENSE) file for more information.\n\n## \ud83d\udce7 Contact\n\nFor any inquiries, feel free to reach out to [dj@codetestcode.io](mailto:dj@codetestcode.io).\n\n---\n\n### \ud83d\udcc8 Project Status\n\nThis project is under active development. New features, bug fixes, and enhancements are regularly added. Stay tuned for updates!\n",
"bugtrack_url": null,
"license": null,
"summary": "A library for executing actions at random times within a specified duration.",
"version": "0.1.0",
"project_urls": {
"Bug Tracker": "https://github.com/copyleftdev/randexcpy/issues",
"Homepage": "https://github.com/copleftdev/randexcpy"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "164fa2a1ea0527fe49b45f6bed4d32f09bb3d1a225752337a5cd2c2dfa12ea9f",
"md5": "f29cca1d37d321e187d22b1eb9fa8e2b",
"sha256": "a4d753cde4ac58b57e27a0a1017062df64a1f2aaecdf82ecc956bf9f45f37924"
},
"downloads": -1,
"filename": "randexcpy-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "f29cca1d37d321e187d22b1eb9fa8e2b",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 7793,
"upload_time": "2024-08-25T19:45:45",
"upload_time_iso_8601": "2024-08-25T19:45:45.782258Z",
"url": "https://files.pythonhosted.org/packages/16/4f/a2a1ea0527fe49b45f6bed4d32f09bb3d1a225752337a5cd2c2dfa12ea9f/randexcpy-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "706b8c041e03ac0f1a5515cfe2a787693867739467be1aeeeeb663d1af57531b",
"md5": "1d78a49e1bfcba208297e45932aa16d0",
"sha256": "db497940d94365a5c0805d551f6df32ab1166d7b3f8b7d4689b1555675ba4ab6"
},
"downloads": -1,
"filename": "randexcpy-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "1d78a49e1bfcba208297e45932aa16d0",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 6887,
"upload_time": "2024-08-25T19:45:47",
"upload_time_iso_8601": "2024-08-25T19:45:47.371581Z",
"url": "https://files.pythonhosted.org/packages/70/6b/8c041e03ac0f1a5515cfe2a787693867739467be1aeeeeb663d1af57531b/randexcpy-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-08-25 19:45:47",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "copleftdev",
"github_project": "randexcpy",
"github_not_found": true,
"lcname": "randexcpy"
}
JSON
