# AsyncGear
Think about such a scene, some object has different states or periods, well we call periods. Among these periods, there
can be only one that could be the present period. For example, a human can only be in one period among baby, youth, adult,
old man and dead.
So now not only can you get the exact period, but also await inside the period, await outside the period, await the
instant when the object enters the period and await the instant when the object exits the period.
Remember to distribute different periods for one object.
## uvloop is highly recommended!
### tip
For convenience, remember that the python codes below are extracted from this wrapper:
```python
import asyncio
# As recommend, use uvloop
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
from AsyncGear import Gear, run_when_enter, run_when_exit, run_when_inside, run_when_outside, when_enter, when_exit, when_inside, when_outside
async def main():
# This is the place where these codes are extracted away.
pass
loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()
# clear
to_cancel = asyncio.all_tasks(loop)
for task in to_cancel:
task.cancel()
loop.run_until_complete(
asyncio.gather(*to_cancel, return_exceptions=True)
)
loop.run_until_complete(loop.shutdown_asyncgens())
```
So if you want to run these codes, you must wrap them back respectively.
Or if you are familiar with asyncio REPL(type 'python -m asyncio' in terminal for python3.8 and above), you can migrate
these codes to run in a terminal.
---
### [Install](#Install) · [Usage](#Usage) ·
---
## Install
[AsyncGear in **PyPI**](https://pypi.org/project/AsyncGear/)
```shell
pip install AsyncGear
```
---
## Usage
This module provides methods as follows:
```python
from AsyncGear import Gear, run_when_enter, run_when_exit, run_when_inside, run_when_outside, when_enter, when_exit, when_inside, when_outside
```
### Gear
Gear is the core usage, which provides an interface for an object to manipulate the bounded gear.
```python
# to be simple, the target object is a string
Gear('Tom').add_periods('sleep', 'awaken') # the first added period would be the default.
Gear('Tom').add_periods('sleepwalking') # add_periods could be dynamic.
# show the present period.
print(Gear('Tom').get_present_period())
# show all possible periods
print(Gear('Tom').get_period_names())
# beforehand wait the target time of the target period
async def wait_enter(gear: Gear, period):
await asyncio.create_task(gear.wait_enter_period(period))
print(f'enter {period}')
asyncio.create_task(wait_enter(Gear('Tom'), 'awaken'))
async def wait_exit(gear: Gear, period):
await asyncio.create_task(gear.wait_exit_period(period))
print(f'exit {period}')
asyncio.create_task(wait_exit(Gear('Tom'), 'sleep'))
async def wait_outside(gear: Gear, period):
while True:
await asyncio.create_task(gear.wait_outside_period(period))
await asyncio.sleep(0.3)
print(f'outside {period}')
asyncio.create_task(wait_outside(Gear('Tom'), 'sleep'))
async def wait_inside(gear: Gear, period):
while True:
await asyncio.create_task(gear.wait_inside_period(period))
await asyncio.sleep(0.3)
print(f'inside {period}')
asyncio.create_task(wait_inside(Gear('Tom'), 'awaken'))
await asyncio.create_task(Gear('Tom').set_period('awaken'))
# stay in awaken for 1 seconds
await asyncio.sleep(1)
loop.stop()
```
Result
```shell
2021-01-05 17:15:23.600 | DEBUG | AsyncGear.AsyncGear:_set_obj_period:74 - set 'Tom' to period sleep.
2021-01-05 17:15:23.601 | DEBUG | AsyncGear.AsyncGear:_set_obj_period:74 - set 'Tom' to period awaken.
2021-01-05 17:15:23.601 | DEBUG | AsyncGear.AsyncGear:_set_obj_period:74 - set 'Tom' to period awaken.
2021-01-05 17:15:23.601 | DEBUG | AsyncGear.AsyncGear:_set_obj_period:74 - set 'Tom' to period awaken.
sleep
dict_keys(['sleep', 'awaken', 'sleepwalking'])
exit sleep
enter awaken
outside sleep
inside awaken
outside sleep
inside awaken
outside sleep
inside awaken
```
### Gear(obj).lock
You could lock the gear period. If locked, the gear could no more be changed the period, or raise PermissionError.
Of course, you could wait the gear unlocked.
```python
import traceback
Gear('Tom').add_periods('sleep', 'awaken')
Gear('Tom').lock()
try:
await asyncio.create_task(Gear('Tom').set_period('awaken'))
except:
print(traceback.format_exc())
async def wait2set_awaken():
await Gear('Tom').wait_unlock()
await asyncio.create_task(Gear('Tom').set_period('awaken'))
loop.stop()
asyncio.create_task(wait2set_awaken())
await asyncio.sleep(1)
Gear('Tom').unlock()
```
```shell
2021-04-13 12:05:06.650 | DEBUG | AsyncGear.AsyncPeriod:filled_slots_num:43 - set 'Tom' to period sleep.
Traceback (most recent call last):
File "<console>", line 2, in <module>
...
raise PermissionError('The gear is locked.')
PermissionError: The gear is locked.
2021-04-13 12:05:07.652 | DEBUG | AsyncGear.AsyncPeriod:filled_slots_num:43 - set 'Tom' to period awaken.
```
### Gear(obj).prev_period
Get the previously set period name, or None.
```python
Gear('Tom').add_periods('sleep', 'awaken') # the first added period would be the default.
print(Gear('Tom').prev_period)
await asyncio.create_task(Gear('Tom').set_period('awaken'))
print(Gear('Tom').prev_period)
loop.stop()
```
Result:
```shell
2021-04-17 18:13:54.257 | DEBUG | AsyncGear.AsyncPeriod:filled_slots_num:45 - set 'Tom' to period sleep.
2021-04-17 18:13:54.258 | DEBUG | AsyncGear.AsyncPeriod:filled_slots_num:45 - set 'Tom' to period awaken.
None
sleep
```
### Gear(obj).current_set_datetime
Get the UTC datetime when the present period is set.
```python
Gear('Tom').add_periods('sleep', 'awaken') # the first added period would be the default.
print(Gear('Tom').current_set_datetime())
await asyncio.sleep(1)
await asyncio.create_task(Gear('Tom').set_period('awaken'))
print(Gear('Tom').current_set_datetime())
loop.stop()
```
Result
```shell
2021-04-17 18:23:02.738 | DEBUG | AsyncGear.AsyncPeriod:filled_slots_num:45 - set 'Tom' to period sleep.
2021-04-17 10:23:02.738709
2021-04-17 10:23:03.741432
2021-04-17 18:23:03.741 | DEBUG | AsyncGear.AsyncPeriod:filled_slots_num:45 - set 'Tom' to period awaken.
```
### Gear(obj).delete
When you no more need a gear, you'd better delete it to save RAM. Especially when you dynamically keep creating new gears, you must keep
deleting the old gears.
```python
Gear('Tom').add_periods('sleep', 'awaken')
Gear('Tom').delete()
```
### run_when_enter
run_when_enter is to decorate a function or coroutine function to be run when just entering the designated period of
the designated object gear.
```python
Gear('Tom').add_periods('sleep', 'awaken')
@run_when_enter('Tom', 'awaken')
def enter_test():
print('synchronous callback')
@run_when_enter('Tom', 'awaken')
async def async_enter_test():
print('asynchronous callback')
await asyncio.create_task(Gear('Tom').set_period('awaken'))
loop.stop()
```
Result
```shell
synchronous callback
asynchronous callback
2021-01-05 17:20:34.636 | DEBUG | AsyncGear.AsyncGear:_set_obj_period:74 - set 'Tom' to period sleep.
2021-01-05 17:20:34.636 | DEBUG | AsyncGear.AsyncGear:_set_obj_period:74 - set 'Tom' to period awaken.
2021-01-05 17:20:34.636 | DEBUG | AsyncGear.AsyncGear:_set_obj_period:74 - set 'Tom' to period awaken.
```
### run_when_exit
run_when_exit is to decorate a function or coroutine function to be run when just exiting the designated period of
the designated object gear.
```python
Gear('Tom').add_periods('sleep', 'awaken')
@run_when_exit('Tom', 'sleep')
def enter_test():
print('synchronous callback')
@run_when_exit('Tom', 'sleep')
async def async_enter_test():
print('asynchronous callback')
await asyncio.create_task(Gear('Tom').set_period('awaken'))
loop.stop()
```
Result
```shell
synchronous callback
asynchronous callback
2021-01-05 17:23:13.993 |
PyPI 官网下载 | AsyncGear-4.5.2.tar.gz
版权申诉
116 浏览量
2022-01-09
05:06:30
上传
评论
收藏 11KB GZ 举报
挣扎的蓝藻
- 粉丝: 13w+
- 资源: 15万+