agronholm / agronholm/apscheduler

Implementing wrapper schedulers

未關閉
#1,085 3 則留言 0 個 reaction 已指派 0 人 在 GitHub 檢視
enhancement
主要語言
Python
星號
7.6k
分支
783
平均合併
4 天 8 小時
30 天內合併 PR
5

描述

### Things to check first

- [x] I have searched the existing issues and didn't find my feature already requested there

### Feature description

I was told by agronholm that this library existed after attempting to write my own implementation of an asynchronous scheduler that utilizes tools for scheduling events in consecutive amounts and is fully customizable with timedeltas for extra percision with timeouts. But I what I really wanted was a way to write a cleanly made interface with the code without needing to pass arguments as often (lazily instead) as the one I had shared in the matrix server after having just finished it all recently. This new version has all the bugs fixed.

So now I have an idea. There is no reason for me to compete at all and I see reason for this concept to get added here some key features to point out are

- Callbacks for start and finishing (takes multiple) (Fully Interchangeable if needed. inspired by aiohttp's TraceConfig object)
- Wrappers for helping keep scheduling neat and organized.
- passable arguments from starting the code to pass along through the entire thing.

### Use case

Here I have a duplicate of said code I shared on the matrix server with a few bugs patched.
https://gist.github.com/Vizonex/f3b1cf2bccff392a26b64c9b62373e19

```python
from anyio_sched import Scheduler, TraceEvent # Know that this was my own mini library
# but the same logic could be utilized with this library. where you have some context variables that could be lazily
# passed after parsing through an argument parser or any user's choice.
from typer import run
from httpx import AsyncClient

sched: Sheduler[str] = Scheduler() # Typehinting was important to me for positional arguments globally :)

trace_event = TraceEvent()

@trace_event.on_start
async def on_start(event, host:str):
print(f"{event.name} has started get request to {host}")

@trace_event.on_finish
async def on_end(event, data:bytes, host:str):
print(f"Got: {data} from {host} after calling {event.name}")

@sched.event(traces=[trace_event])
async def simple_get_request(host:str):
async with AsyncClient() as client:
resp = await client.get(host)
data = await resp.aread()
return data

def main(host: str):
return sched.start(host, backend='asyncio')

if __name__ == "__main__":
run(main)
```

貢獻指南

開啟貢獻指南

評估

這個 Issue 還沒有評估資料。

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。