Schedule asyncio coroutines
Project description
everytime - Schedule asyncio coroutines
TLDR
@every.other.wednesday.at(hour=12)
async def do_something():
...
Full Example
from everytime import every
import everytime
@every(5).seconds
async def greet():
print("Hello")
everytime.run_forever()
How to schedule coroutines
everytime expressions as decorators
All everytime expressions can be used as function decorators.
@every(5).seconds
async def greet():
print("Hello")
@schedule
Aternatively you can wrap the everytime expression into a call to @schedule
.
@schedule(every(5).seconds)
async def greet():
print("Hello")
This allows you to pass custom datetime iterables to @schedule
(see Schedule custom times).
Schedule custom times
@schedule
accepts datetime iterables. The following schedules work:
@schedule([datetime.fromisoformat('2022-11-01T12:00:00'), datetime.fromisoformat('2023-01-01T12:00:00')])
@schedule(itertools.islice(every.day, 5))
@schedule(map(lambda _: datetime.now() + timedelta(seconds=1), sys.stdin))
do()
If you prefer to keep your function definitions and scheduling rules separate, use the do
-function.
async def greet():
print("Hello")
every(5).seconds.do(greet)
Supported Expressions
Quantification
Every time unit can be quantified by every
, every.other
or every(n)
:
every.second
every.other.second
every(5).seconds
Supported time units
The supported time units are
millisecond
second
minute
hour
day
week
Weekdays
Also, weekdays monday
through sunday
are supported. every.wednesday
starts on the next Wednesday. If today is a Wednesday, every.wednesday
starts today.
Specific time of the day
day
and the weekdays can be scheduled for a specific time of the day:
every.day.at(hour=12, minute=15)
(Note that hour
is 24-hour based)
Event Loops
everytime uses asyncio
and schedules coroutines on an event loop.
Default Behavior
By default, all coroutines are scheduled on the same event loop. After all schedules are set, the loop must be invoked with everytime.run_forever()
@schedule(every.second)
async def greet():
print("Hello")
everytime.run_forever()
Async Environment
If called in an async environment (i.e. there is already an event loop running), coroutines are scheduled on asyncio.get_running_loop()
.
async def main():
@schedule(every.second)
async def greet():
print("Hello")
await asyncio.sleep(10)
asyncio.run(main())
Note, that the scheduling only works while the loop is running. In this case, greet
will only be called every second, while main
is still running.
Custom Event Loops
You can schedule your coroutines to run on a custom event loop by passing an optional argument loop
to@schedule
or do()
.
l = asyncio.new_event_loop()
@schedule(every.second, loop=l)
async def greet():
print("Hello")
l.run_forever()
l = asyncio.new_event_loop()
async def greet():
print("Hello")
every.second.do(greet, loop=l)
l.run_forever()
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for everytime-0.2.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | aaef583220f66e0fbe5e1df809bf17aa86826044c0f2ff2c9ea0dd0842142199 |
|
MD5 | 7f50a7f4b1e9b266de7606fda3cdd588 |
|
BLAKE2b-256 | e9e358f1295bfb720cfff7a42783a4d6e47e7f606bbb68fb56b42a449b2ad0ec |