| Name | looperation JSON |
| Version |
2.0.0
JSON |
| download |
| home_page | https://github.com/Shahaf-F-S/looperator |
| Summary | A python module to easily run loop based operations, control the starting, stopping of the loop, and handle exceptions in real time. |
| upload_time | 2024-01-20 12:51:59 |
| maintainer | |
| docs_url | None |
| author | Shahaf Frank-Shapir |
| requires_python | |
| license | MIT |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# looperator
> A python module to easily run loop based or condition-based operations, control the starting, pausing, unpausing, timeout, delay, and stopping of the loop, and handle exceptions in real time.
Installation
-----------
````
pip install looperator
````
example
-----------
* all attributes of the operator are being evaluated at runtime, thus any change for any attribute during runtime is valid.
* when pausing the operator or superator, the timeout and stopping processes are paused as well.
create an operator object
````python
import time
import random
import datetime as dt
from looperation import Operator
operator = Operator(
operation=lambda value: print(value),
args_collector=lambda: (random.randint(0, 10),),
delay=dt.timedelta(seconds=1)
)
````
run the operation with timeout pause adter some time
````python
print("starting process, 10 seconds timeout")
operator.run(timeout=dt.timedelta(seconds=10))
time.sleep(5)
operator.pause()
print("passed after 5 seconds")
print("paused for 5 more seconds")
````
output
````
starting process, 10 seconds timeout
10
6
2
7
0
passed after 5 seconds
paused for 5 more seconds
````
wait some time while paused and continue
````python
time.sleep(5)
print("5 seconds left")
operator.unpause()
time.sleep(5)
print("process ending")
````
output
````
5 seconds left
5
1
4
4
9
process ending
````
for exception handling pass to the operator object a handler object
````python
from looperation import Handler, Operator
handler = Handler(
exceptions=[ZeroDivisionError],
exception_handler=print,
exception_callback=lambda h: print("error")
)
operator = Operator(
operation=lambda value: print(1 / value),
args_collector=lambda: (0,),
delay=dt.timedelta(seconds=1),
handler=handler
)
operator.run()
````
output
````
error
division by zero
````
run an operation as long as a condition is met
````python
import time
import random
import datetime as dt
from looperation import Operator
conditions = [False, False, False]
operator = Operator(
operation=lambda value: print(value),
args_collector=lambda: (random.randint(0, 11),),
stopping_collector=lambda: any(conditions),
delay=dt.timedelta(seconds=1),
loop_stopping=True
)
# loop_stopping=True for evaluating the stopping_collector callable during the operation loop,
# loop_stopping=False to evaluate in a different thread.
print("starting process")
operator.run()
time.sleep(5)
conditions.append(True)
print("process ending")
````
output
````
starting process
2
7
4
6
8
process ending
````
run a loop operation as long as a condition is met
````python
import time
import random
import datetime as dt
from looperation import Operator
def loop_operation(operation: Operator):
while operation.operating:
print(random.randint(0, 11))
operation.start_waiting(operation.delay)
conditions = [False, False, False]
operator = Operator(
stopping_collector=lambda: any(conditions),
delay=dt.timedelta(seconds=1),
loop=False # for calling the operation callable only once.
)
operator.operation = lambda: loop_operation(operator)
print("starting process")
operator.run()
time.sleep(5)
conditions.append(True)
print("process ending")
````
output
````
starting process
2
9
7
9
8
process ending
````
run the operation in the main thread
````python
import random
import datetime as dt
from looperation import Operator
operator = Operator(
operation=lambda value: print(value),
args_collector=lambda: (random.randint(0, 11),),
delay=dt.timedelta(seconds=1),
timeout=dt.timedelta(seconds=5)
)
print("starting process")
operator.run(block=True)
print("process ending")
````
output
````
starting process
8
9
5
2
3
process ending
````
Operator constructor signature
````python
TimeDuration = float | dt.timedelta
TimeDestination = TimeDuration | dt.datetime
def __init__(
self,
name: str = None, *,
operation: Callable[..., _O] = None,
args_collector: Callable[[], Iterable[Any]] = None,
kwargs_collector: Callable[[], dict[str, Any]] = None,
stopping_collector: Callable[[], bool] = None,
termination: Callable[[], Any] = None,
handler: Handler = None,
loop: bool = True,
warn: bool = False,
loop_stopping: bool = None,
delay: TimeDuration = None,
block: bool = False,
coroutine: bool = False,
wait: float | TimeDestination = None,
timeout: float | TimeDestination = None
)
````
.run method signature
````python
TimeDuration = float | dt.timedelta
TimeDestination = TimeDuration | dt.datetime
def run(
self,
loop: bool = None,
loop_stopping: bool = None,
block: bool = None,
wait: TimeDestination = None,
timeout: TimeDestination = None
)
````
all main methods of the Operator class
```python
operator.run()
operator.start_timeout(duration=dt.timedelta(seconds=5))
operator.stop_timeout()
operator.pause()
operator.unpause()
operator.start_stopping()
operator.stop_stopping()
operator.start_waiting(duration=dt.timedelta(seconds=5))
operator.stop()
```
Using a Superator object - an Operator of Operators
````python
import time
import datetime as dt
import random
from looperation import Operator, Superator
operator1 = Operator(
operation=lambda value: print(value, "operator 1"),
args_collector=lambda: (random.randint(0, 5),),
delay=dt.timedelta(seconds=1)
)
operator2 = Operator(
operation=lambda value: print(value, "operator 2"),
args_collector=lambda: (random.randint(5, 10),),
delay=dt.timedelta(seconds=1)
)
superator = Superator(
operators=[operator1, operator2]
)
print("starting process, 10 seconds timeout")
superator.run(timeout=dt.timedelta(seconds=4))
time.sleep(2)
superator.pause()
print("passed after 5 seconds")
print("paused for 5 more seconds")
time.sleep(2)
print("5 seconds left")
superator.unpause()
time.sleep(2)
print("process ending")
````
output
````
starting process, 10 seconds timeout
0 operator 1
9 operator 2
2 operator 1
7 operator 2
passed after 5 seconds
paused for 5 more seconds
5 seconds left
2 operator 1
6 operator 2
2 operator 1
7 operator 2
process ending
````
Superator constructor signature
````python
TimeDuration = float | dt.timedelta
TimeDestination = TimeDuration | dt.datetime
def __init__(
operators: Iterable[Operator],
handler: Handler = None,
stopping_collector: Callable[[], bool] = None,
termination: Callable[[], Any] = None,
delay: TimeDuration = None,
block: bool = False,
wait: TimeDestination = None,
timeout: TimeDestination = None
)
````
.run method signature
````python
TimeDuration = float | dt.timedelta
TimeDestination = TimeDuration | dt.datetime
def run(
self,
loop: bool = None,
loop_stopping: bool = None,
block: bool = None,
wait: TimeDestination = None,
timeout: TimeDestination = None
)
````
all main methods of the Operator class
```python
CONTROL_ALL_OPERATORS = True # when False, doesn't affect the operators.
superator.run()
superator.start_timeout(duration=dt.timedelta(seconds=5))
superator.stop_timeout()
superator.pause(operators=CONTROL_ALL_OPERATORS)
superator.unpause(operators=CONTROL_ALL_OPERATORS)
superator.start_stopping()
superator.stop_stopping()
superator.start_waiting(duration=dt.timedelta(seconds=5))
superator.stop(operators=CONTROL_ALL_OPERATORS)
```
Raw data
{
"_id": null,
"home_page": "https://github.com/Shahaf-F-S/looperator",
"name": "looperation",
"maintainer": "",
"docs_url": null,
"requires_python": "",
"maintainer_email": "",
"keywords": "",
"author": "Shahaf Frank-Shapir",
"author_email": "shahaffrs@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/93/2f/e5ffadb907b88cbedb9941e5cc442b6de3124dd23c763385ebae947903ee/looperation-2.0.0.tar.gz",
"platform": null,
"description": "# looperator\r\n\r\n> A python module to easily run loop based or condition-based operations, control the starting, pausing, unpausing, timeout, delay, and stopping of the loop, and handle exceptions in real time.\r\n\r\n\r\nInstallation\r\n-----------\r\n````\r\npip install looperator\r\n````\r\n\r\nexample\r\n-----------\r\n* all attributes of the operator are being evaluated at runtime, thus any change for any attribute during runtime is valid.\r\n* when pausing the operator or superator, the timeout and stopping processes are paused as well.\r\n\r\ncreate an operator object\r\n\r\n````python\r\nimport time\r\nimport random\r\nimport datetime as dt\r\n\r\nfrom looperation import Operator\r\n\r\noperator = Operator(\r\n operation=lambda value: print(value),\r\n args_collector=lambda: (random.randint(0, 10),),\r\n delay=dt.timedelta(seconds=1)\r\n)\r\n````\r\n\r\nrun the operation with timeout pause adter some time\r\n````python\r\nprint(\"starting process, 10 seconds timeout\")\r\n\r\noperator.run(timeout=dt.timedelta(seconds=10))\r\n\r\ntime.sleep(5)\r\n\r\noperator.pause()\r\n\r\nprint(\"passed after 5 seconds\")\r\nprint(\"paused for 5 more seconds\")\r\n````\r\n\r\noutput\r\n````\r\nstarting process, 10 seconds timeout\r\n10\r\n6\r\n2\r\n7\r\n0\r\npassed after 5 seconds\r\npaused for 5 more seconds\r\n````\r\n\r\nwait some time while paused and continue\r\n````python\r\ntime.sleep(5)\r\n\r\nprint(\"5 seconds left\")\r\n\r\noperator.unpause()\r\n\r\ntime.sleep(5)\r\n\r\nprint(\"process ending\")\r\n````\r\n\r\noutput\r\n````\r\n5 seconds left\r\n5\r\n1\r\n4\r\n4\r\n9\r\nprocess ending\r\n````\r\n\r\nfor exception handling pass to the operator object a handler object\r\n\r\n````python\r\nfrom looperation import Handler, Operator\r\n\r\nhandler = Handler(\r\n exceptions=[ZeroDivisionError],\r\n exception_handler=print,\r\n exception_callback=lambda h: print(\"error\")\r\n)\r\n\r\noperator = Operator(\r\n operation=lambda value: print(1 / value),\r\n args_collector=lambda: (0,),\r\n delay=dt.timedelta(seconds=1),\r\n handler=handler\r\n)\r\n\r\noperator.run()\r\n````\r\n\r\noutput\r\n````\r\nerror\r\ndivision by zero\r\n````\r\n\r\nrun an operation as long as a condition is met\r\n\r\n````python\r\nimport time\r\nimport random\r\nimport datetime as dt\r\n\r\nfrom looperation import Operator\r\n\r\nconditions = [False, False, False]\r\n\r\noperator = Operator(\r\n operation=lambda value: print(value),\r\n args_collector=lambda: (random.randint(0, 11),),\r\n stopping_collector=lambda: any(conditions),\r\n delay=dt.timedelta(seconds=1),\r\n loop_stopping=True\r\n)\r\n\r\n# loop_stopping=True for evaluating the stopping_collector callable during the operation loop, \r\n# loop_stopping=False to evaluate in a different thread.\r\n\r\nprint(\"starting process\")\r\n\r\noperator.run()\r\n\r\ntime.sleep(5)\r\n\r\nconditions.append(True)\r\n\r\nprint(\"process ending\")\r\n````\r\n\r\noutput\r\n````\r\nstarting process\r\n2\r\n7\r\n4\r\n6\r\n8\r\nprocess ending\r\n````\r\n\r\nrun a loop operation as long as a condition is met\r\n\r\n````python\r\nimport time\r\nimport random\r\nimport datetime as dt\r\n\r\nfrom looperation import Operator\r\n\r\n\r\ndef loop_operation(operation: Operator):\r\n while operation.operating:\r\n print(random.randint(0, 11))\r\n\r\n operation.start_waiting(operation.delay)\r\n\r\n\r\nconditions = [False, False, False]\r\n\r\noperator = Operator(\r\n stopping_collector=lambda: any(conditions),\r\n delay=dt.timedelta(seconds=1),\r\n loop=False # for calling the operation callable only once.\r\n)\r\n\r\noperator.operation = lambda: loop_operation(operator)\r\n\r\nprint(\"starting process\")\r\n\r\noperator.run()\r\n\r\ntime.sleep(5)\r\n\r\nconditions.append(True)\r\n\r\nprint(\"process ending\")\r\n````\r\n\r\noutput\r\n````\r\nstarting process\r\n2\r\n9\r\n7\r\n9\r\n8\r\nprocess ending\r\n````\r\n\r\nrun the operation in the main thread\r\n\r\n````python\r\nimport random\r\nimport datetime as dt\r\n\r\nfrom looperation import Operator\r\n\r\noperator = Operator(\r\n operation=lambda value: print(value),\r\n args_collector=lambda: (random.randint(0, 11),),\r\n delay=dt.timedelta(seconds=1),\r\n timeout=dt.timedelta(seconds=5)\r\n)\r\n\r\nprint(\"starting process\")\r\n\r\noperator.run(block=True)\r\n\r\nprint(\"process ending\")\r\n````\r\n\r\noutput\r\n````\r\nstarting process\r\n8\r\n9\r\n5\r\n2\r\n3\r\nprocess ending\r\n````\r\n\r\nOperator constructor signature\r\n````python\r\nTimeDuration = float | dt.timedelta\r\nTimeDestination = TimeDuration | dt.datetime\r\n\r\ndef __init__(\r\n self,\r\n name: str = None, *,\r\n operation: Callable[..., _O] = None,\r\n args_collector: Callable[[], Iterable[Any]] = None,\r\n kwargs_collector: Callable[[], dict[str, Any]] = None,\r\n stopping_collector: Callable[[], bool] = None,\r\n termination: Callable[[], Any] = None,\r\n handler: Handler = None,\r\n loop: bool = True,\r\n warn: bool = False,\r\n loop_stopping: bool = None,\r\n delay: TimeDuration = None,\r\n block: bool = False,\r\n coroutine: bool = False,\r\n wait: float | TimeDestination = None,\r\n timeout: float | TimeDestination = None\r\n)\r\n````\r\n\r\n.run method signature\r\n````python\r\nTimeDuration = float | dt.timedelta\r\nTimeDestination = TimeDuration | dt.datetime\r\n\r\ndef run(\r\n self,\r\n loop: bool = None,\r\n loop_stopping: bool = None,\r\n block: bool = None,\r\n wait: TimeDestination = None,\r\n timeout: TimeDestination = None\r\n)\r\n````\r\n\r\nall main methods of the Operator class\r\n```python\r\noperator.run()\r\noperator.start_timeout(duration=dt.timedelta(seconds=5))\r\noperator.stop_timeout()\r\noperator.pause()\r\noperator.unpause()\r\noperator.start_stopping()\r\noperator.stop_stopping()\r\noperator.start_waiting(duration=dt.timedelta(seconds=5))\r\noperator.stop()\r\n```\r\n\r\nUsing a Superator object - an Operator of Operators\r\n\r\n````python\r\nimport time\r\nimport datetime as dt\r\nimport random\r\n\r\nfrom looperation import Operator, Superator\r\n\r\noperator1 = Operator(\r\n operation=lambda value: print(value, \"operator 1\"),\r\n args_collector=lambda: (random.randint(0, 5),),\r\n delay=dt.timedelta(seconds=1)\r\n)\r\noperator2 = Operator(\r\n operation=lambda value: print(value, \"operator 2\"),\r\n args_collector=lambda: (random.randint(5, 10),),\r\n delay=dt.timedelta(seconds=1)\r\n)\r\n\r\nsuperator = Superator(\r\n operators=[operator1, operator2]\r\n)\r\n\r\nprint(\"starting process, 10 seconds timeout\")\r\n\r\nsuperator.run(timeout=dt.timedelta(seconds=4))\r\n\r\ntime.sleep(2)\r\n\r\nsuperator.pause()\r\n\r\nprint(\"passed after 5 seconds\")\r\nprint(\"paused for 5 more seconds\")\r\n\r\ntime.sleep(2)\r\n\r\nprint(\"5 seconds left\")\r\n\r\nsuperator.unpause()\r\n\r\ntime.sleep(2)\r\n\r\nprint(\"process ending\")\r\n````\r\n\r\noutput\r\n````\r\nstarting process, 10 seconds timeout\r\n0 operator 1\r\n9 operator 2\r\n2 operator 1\r\n7 operator 2\r\npassed after 5 seconds\r\npaused for 5 more seconds\r\n5 seconds left\r\n2 operator 1\r\n6 operator 2\r\n2 operator 1\r\n7 operator 2\r\nprocess ending\r\n````\r\n\r\nSuperator constructor signature\r\n````python\r\nTimeDuration = float | dt.timedelta\r\nTimeDestination = TimeDuration | dt.datetime\r\n\r\ndef __init__(\r\n operators: Iterable[Operator],\r\n handler: Handler = None,\r\n stopping_collector: Callable[[], bool] = None,\r\n termination: Callable[[], Any] = None,\r\n delay: TimeDuration = None,\r\n block: bool = False,\r\n wait: TimeDestination = None,\r\n timeout: TimeDestination = None\r\n)\r\n````\r\n\r\n.run method signature\r\n````python\r\nTimeDuration = float | dt.timedelta\r\nTimeDestination = TimeDuration | dt.datetime\r\n\r\ndef run(\r\n self,\r\n loop: bool = None,\r\n loop_stopping: bool = None,\r\n block: bool = None,\r\n wait: TimeDestination = None,\r\n timeout: TimeDestination = None\r\n)\r\n````\r\n\r\nall main methods of the Operator class\r\n```python\r\n\r\nCONTROL_ALL_OPERATORS = True # when False, doesn't affect the operators.\r\n\r\nsuperator.run()\r\nsuperator.start_timeout(duration=dt.timedelta(seconds=5))\r\nsuperator.stop_timeout()\r\nsuperator.pause(operators=CONTROL_ALL_OPERATORS)\r\nsuperator.unpause(operators=CONTROL_ALL_OPERATORS)\r\nsuperator.start_stopping()\r\nsuperator.stop_stopping()\r\nsuperator.start_waiting(duration=dt.timedelta(seconds=5))\r\nsuperator.stop(operators=CONTROL_ALL_OPERATORS)\r\n```\r\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A python module to easily run loop based operations, control the starting, stopping of the loop, and handle exceptions in real time.",
"version": "2.0.0",
"project_urls": {
"Homepage": "https://github.com/Shahaf-F-S/looperator"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "932fe5ffadb907b88cbedb9941e5cc442b6de3124dd23c763385ebae947903ee",
"md5": "fb80704b126573652d671458b8b00ccd",
"sha256": "e1c027a9ad7668f32fc8ecdc01d06ad339a1ac405f6953cbe2700eafd7abc0b1"
},
"downloads": -1,
"filename": "looperation-2.0.0.tar.gz",
"has_sig": false,
"md5_digest": "fb80704b126573652d671458b8b00ccd",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 12631,
"upload_time": "2024-01-20T12:51:59",
"upload_time_iso_8601": "2024-01-20T12:51:59.528447Z",
"url": "https://files.pythonhosted.org/packages/93/2f/e5ffadb907b88cbedb9941e5cc442b6de3124dd23c763385ebae947903ee/looperation-2.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-01-20 12:51:59",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Shahaf-F-S",
"github_project": "looperator",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [],
"lcname": "looperation"
}
