-
Notifications
You must be signed in to change notification settings - Fork 749
Description
Things to check first
- 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
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)