Skip to main content

An asyncio-based scheduling framework designed for execution of periodic task with integrated support for dependency injection, enabling efficient and flexible task management

Project description

aioclock

Release Build status Commit activity License

An asyncio-based scheduling framework designed for execution of periodic task with integrated support for dependency injection, enabling efficient and flexiable task management

Features

Aioclock offers:

  • Async: 100% Async, very light, fast and resource friendly
  • Scheduling: Keep scheduling tasks for you
  • Group: Group your task, for better code maintainability
  • Trigger: Already defined and easily extendable triggers, to trigger your scheduler to be started
  • Easy syntax: Library's syntax is very easy and enjoyable, no confusing hierarchy
  • Pydantic v2 validation: Validate all your trigger on startup using pydantic 2. Fastest to fail possible!
  • Soon: Running the task dispatcher (scheduler) on different process by default, so CPU intensive stuff on task won't delay the scheduling
  • Soon: Backend support, to allow horizontal scalling, by synchronizing, maybe using Redis

Getting started

To Install aioclock, simply do

pip install aioclock

Help

See documentation for more details.

A Sample Example

import asyncio

from aioclock import AioClock, At, Depends, Every, Forever, Once, OnShutDown, OnStartUp
from aioclock.group import Group

# groups.py
group = Group()


def more_useless_than_me():
    return "I'm a dependency. I'm more useless than a screen door on a submarine."


@group.task(trigger=Every(seconds=10))
async def every():
    print("Every 10 seconds, I make a quantum leap. Where will I land next?")


@group.task(trigger=Every(seconds=5))
def even_sync_works():
    print("I'm a synchronous task. I work even in async world.")


@group.task(trigger=At(tz="UTC", hour=0, minute=0, second=0))
async def at():
    print(
        "When the clock strikes midnight... I turn into a pumpkin. Just kidding, I run this task!"
    )


@group.task(trigger=Forever())
async def forever(val: str = Depends(more_useless_than_me)):
    await asyncio.sleep(2)
    print("Heartbeat detected. Still not a zombie. Will check again in a bit.")
    assert val == "I'm a dependency. I'm more useless than a screen door on a submarine."


@group.task(trigger=Once())
async def once():
    print("Just once, I get to say something. Here it goes... I love lamp.")


# app.py
app = AioClock()
app.include_group(group)


@app.task(trigger=OnStartUp())
async def startup():
    print(
        "Welcome to the Async Chronicles! Did you know a group of unicorns is called a blessing? Well, now you do!"
    )


@app.task(trigger=OnShutDown())
async def shutdown():
    print("Going offline. Remember, if your code is running, you better go catch it!")


# main.py
if __name__ == "__main__":
    asyncio.run(app.serve())

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aioclock-0.3.0.tar.gz (1.2 MB view details)

Uploaded Source

Built Distribution

aioclock-0.3.0-py3-none-any.whl (19.6 kB view details)

Uploaded Python 3

File details

Details for the file aioclock-0.3.0.tar.gz.

File metadata

  • Download URL: aioclock-0.3.0.tar.gz
  • Upload date:
  • Size: 1.2 MB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.12.4

File hashes

Hashes for aioclock-0.3.0.tar.gz
Algorithm Hash digest
SHA256 ed992fc3e426c769e59feb0860b4e4a17f126471cc518c1d37dccbfa248d2134
MD5 035d9a5538639da539585c53d45a6ee6
BLAKE2b-256 9b586960028c9528c2b9fcd5fc45beadcc9e705d6009551b86b4331c51478070

See more details on using hashes here.

File details

Details for the file aioclock-0.3.0-py3-none-any.whl.

File metadata

  • Download URL: aioclock-0.3.0-py3-none-any.whl
  • Upload date:
  • Size: 19.6 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.12.4

File hashes

Hashes for aioclock-0.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 3cd99b5e06d9a994696205658e574688331c64f017571a0b5410a5a7087be99e
MD5 c29c721eed4c7ee5320834eebe29a3d2
BLAKE2b-256 9fedf3cd1ad1fda9358dd9864fd4beea7350f35b39160b88a7fea8ff0657fed9

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page