calendar-queue

A pure-python calendar queue and library for scheduling events with asyncio.

Many other libraries allow for event/job scheduling but either they are not asynchronous or they are just fire-and-forget. calendar-queue has two purposes:

1. providing the primitives for creating your own scheduler by exporting a CalendarQueue class in which events can be awaited
2. providing a higher level abstraction Calendar class that simplifies the usage of CalendarQueue.

Depending on your needs, you can use one of the two to develop your own event manager/scheduler.

The idea is: we take care of emitting events at the right time, you write the logic for acting accordingly.

Install

pip install calendar-queue

Usage

An example usage of CalendarQueue

import asyncio
from datetime import datetime
from random import randrange
from secrets import token_hex

from calendar_queue import CalendarQueue

# (optional) define the item type 
CustomItem = tuple[str]

# use the low level calendar queue
cq: CalendarQueue[CustomItem] = CalendarQueue()

async def put_random():

    print("Putting new items in the calendar queue. Hit CTRL + C to stop.")

    while True:

        # sleep for a while, just to release the task
        await asyncio.sleep(1)

        scheduled_ts = datetime.now().timestamp() + randrange(1, 5)

        s = token_hex(8)

        current_item: CustomItem = (s)

        print(f"{datetime.now().isoformat()}: putting {current_item} scheduled for {datetime.fromtimestamp(scheduled_ts).isoformat()}")

        cq.put_nowait((scheduled_ts, current_item))


async def get_from_queue():

    while True:
        ts, el = await cq.get()

        print(f"{datetime.now().isoformat()}: getting {el} scheduled for {datetime.fromtimestamp(ts).isoformat()}")

async def main():

    await asyncio.gather(
        asyncio.create_task(put_random()),
        asyncio.create_task(get_from_queue()),
    )


if __name__ == "__main__":
    asyncio.run(main())    

Development

This library is developed using Python 3.11 and pdm as dependency manager.

Testing is done via github actions and it's done on python versions 3.10, 3.11, 3.12 and on latest ubuntu, macos, windows OSes.

For local development you'll need to have pdm installed, then you can:

1. run pdm install to create the virtual environment and install the dependencies
2. run pdm venv activate to activate the virtual environment
3. run the tests using pytest
4. run the linter pylint src

Contributing

Contributions are welcome! Especially for new tests and for improving documentation and examples.

License

The code in this project is released under the MIT License.