Skip to content

AioClock Application

To initialize the AioClock instance, you need to import the AioClock class from the aioclock module. AioClock class represent the aioclock, and handle the tasks and groups that will be run by the aioclock.

Another way to modulize your code is to use Group which is kinda the same idea as router in web frameworks.

AioClock 

AioClock(
    *,
    lifespan: Optional[
        Callable[
            [AioClock],
            AsyncContextManager[AioClock]
            | ContextManager[AioClock],
        ]
    ] = None,
    limiter: Optional[CapacityLimiter] = None
)

AioClock is the main class that will be used to run the tasks. It will be responsible for running the tasks in the right order.

Example 
from aioclock import AioClock, Once
app = AioClock()

@app.task(trigger=Once())
async def main():
    print("Hello World")

To run the aioclock final app simply do:

Example 
from aioclock import AioClock, Once
import asyncio

app = AioClock()

# whatever next comes here
asyncio.run(app.serve())

Lifespan

You can define this startup and shutdown logic using the lifespan parameter of the AioClock instance. It should be as an AsyncContextManager which get AioClock application as arguement. You can find the example below.

Example 
    import asyncio
    from contextlib import asynccontextmanager

    from aioclock import AioClock

    ML_MODEL = [] # just some imaginary component that needs to be started and stopped


    @asynccontextmanager
    async def lifespan(app: AioClock):
        ML_MODEL.append(2)
        print("UP!")
        yield app
        ML_MODEL.clear()
        print("DOWN!")


    app = AioClock(lifespan=lifespan)


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

Here we are simulating the expensive startup operation of loading the model by putting the (fake) model function in the dictionary with machine learning models before the yield. This code will be executed before the application starts operationg, during the startup.

And then, right after the yield, we unload the model. This code will be executed after the application finishes handling requests, right before the shutdown. This could, for example, release resources like memory, a GPU or some database connection.

It would also happen when you're stopping your application gracefully, for example, when you're shutting down your container.

Lifespan could also be synchronus context manager. Check the example below.

Example 
    from contextlib import contextmanager

    from aioclock import AioClock

    ML_MODEL = []

    @contextmanager
    def lifespan_sync(sync_app: AioClock):
        ML_MODEL.append(2)
        print("UP!")
        yield sync_app
        ML_MODEL.clear()
        print("DOWN!")

    sync_app = AioClock(lifespan=lifespan_sync)

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

Attributes:

Name Type Description
lifespan

A context manager that will be used to handle the startup and shutdown of the application. If not provided, the application will run without any startup and shutdown logic. To understand it better, check the examples and documentation above.
limiter

Anyio CapacityLimiter. capacity limiter to use to limit the total amount of threads running Limiter that will be used to limit the number of tasks that are running at the same time. If not provided, it will fallback to the default limiter set on Application level. If no limiter is set on Application level, it will fallback to the default limiter set by AnyIO.
Source code in aioclock/app.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def __init__(
    self,
    *,
    lifespan: Optional[
        Callable[[AioClock], AsyncContextManager[AioClock] | ContextManager[AioClock]]
    ] = None,
    limiter: Optional[anyio.CapacityLimiter] = None,
):
    """
    Initialize AioClock instance.
    No parameters are needed.

    Attributes:
        lifespan:
            A context manager that will be used to handle the startup and shutdown of the application.
            If not provided, the application will run without any startup and shutdown logic.
            To understand it better, check the examples and documentation above.

        limiter:
            Anyio CapacityLimiter. capacity limiter to use to limit the total amount of threads running
            Limiter that will be used to limit the number of tasks that are running at the same time.
            If not provided, it will fallback to the default limiter set on Application level.
            If no limiter is set on Application level, it will fallback to the default limiter set by AnyIO.

    """
    self._groups: list[Group] = []
    self._app_tasks: list[Task] = []
    self._limiter = limiter
    self.lifespan = lifespan

dependencies property 

dependencies

Dependencies provider that will be used to inject dependencies in tasks.

override_dependencies 

override_dependencies(
    original: Callable[..., Any],
    override: Callable[..., Any],
) -> None

Override a dependency with a new one.

Parameters:

Name Type Description Default
original Callable[..., Any]

Original dependency that will be overridden.

 required
override Callable[..., Any]

New dependency that will override the original one.

 required
Example 
from aioclock import AioClock

def original_dependency():
    return 1

def new_dependency():
    return 2

app = AioClock()
app.override_dependencies(original=original_dependency, override=new_dependency)
Source code in aioclock/app.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def override_dependencies(
    self, original: Callable[..., Any], override: Callable[..., Any]
) -> None:
    """Override a dependency with a new one.

    params:
        original:
            Original dependency that will be overridden.
        override:
            New dependency that will override the original one.

    Example:
        ```python
        from aioclock import AioClock

        def original_dependency():
            return 1

        def new_dependency():
            return 2

        app = AioClock()
        app.override_dependencies(original=original_dependency, override=new_dependency)
        ```

    """
    self.dependencies.override(original, override)

include_group 

include_group(group: Group) -> None

Include a group of tasks that will be run by AioClock.

Parameters:

Name Type Description Default
group Group

Group of tasks that will be run together.

 required
Example 
from aioclock import AioClock, Group, Once

app = AioClock()

group = Group()
@group.task(trigger=Once())
async def main():
    print("Hello World")

app.include_group(group)
Source code in aioclock/app.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def include_group(self, group: Group) -> None:
    """Include a group of tasks that will be run by AioClock.

    params:
        group:
            Group of tasks that will be run together.

    Example:
        ```python
        from aioclock import AioClock, Group, Once

        app = AioClock()

        group = Group()
        @group.task(trigger=Once())
        async def main():
            print("Hello World")

        app.include_group(group)
        ```
    """
    self._groups.append(group)
    return None

task 

task(*, trigger: BaseTrigger)

Decorator to add a task to the AioClock instance. If decorated function is sync, aioclock will run it in a thread pool executor, using AnyIO. But if you try to run the decorated function, it will run in the same thread, blocking the event loop. It is intended to not change all your sync functions to coroutine functions, and they can be used outside of aioclock, if needed.

Parameters:

Name Type Description Default
trigger BaseTrigger

BaseTrigger Trigger that will trigger the task to be running.

 required
Example 
from aioclock import AioClock, Once

app = AioClock()

@app.task(trigger=Once())
async def main():
    print("Hello World")
Source code in aioclock/app.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def task(self, *, trigger: BaseTrigger):
    """
    Decorator to add a task to the AioClock instance.
    If decorated function is sync, aioclock will run it in a thread pool executor, using AnyIO.
    But if you try to run the decorated function, it will run in the same thread, blocking the event loop.
    It is intended to not change all your `sync functions` to coroutine functions,
        and they can be used outside of aioclock, if needed.

    params:
        trigger: BaseTrigger
            Trigger that will trigger the task to be running.

    Example:
        ```python

        from aioclock import AioClock, Once

        app = AioClock()

        @app.task(trigger=Once())
        async def main():
            print("Hello World")
        ```
    """

    def decorator(func):
        @wraps(func)
        async def wrapped_funciton(*args, **kwargs):
            if asyncio.iscoroutinefunction(func):
                return await func(*args, **kwargs)
            else:  # run in threadpool to make sure it's not blocking the event loop
                return await asyncify(func, limiter=self._limiter)(*args, **kwargs)

        self._app_tasks.append(
            Task(
                func=inject(wrapped_funciton, dependency_overrides_provider=get_provider()),
                trigger=trigger,
            )
        )
        if asyncio.iscoroutinefunction(func):
            return wrapped_funciton
        else:

            @wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)

            return wrapper

    return decorator

serve async 

serve() -> None

Serves AioClock Run the tasks in the right order. First, run the startup tasks, then run the tasks, and finally run the shutdown tasks.

Source code in aioclock/app.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
async def serve(self) -> None:
    """
    Serves AioClock
    Run the tasks in the right order.
    First, run the startup tasks, then run the tasks, and finally run the shutdown tasks.
    """
    group = Group()
    group._tasks = self._app_tasks
    self.include_group(group)

    if self.lifespan is None:
        await self._run_tasks()
        return

    ctx = self.lifespan(self)

    if isinstance(ctx, AsyncContextManager):
        async with ctx:
            await self._run_tasks()

    elif isinstance(ctx, ContextManager):
        with ctx:
            await self._run_tasks()

    else:
        assert_never(ctx)

Group 

Group(*, limiter: Optional[CapacityLimiter] = None)

Best use case is to have a good modularity and separation of concerns. For example, you can have a group of tasks that are responsible for sending emails. And another group of tasks that are responsible for sending notifications.

Parameters:

Name Type Description Default
limiter Optional[CapacityLimiter]

Anyio CapacityLimiter. capacity limiter to use to limit the total amount of threads running Limiter that will be used to limit the number of tasks that are running at the same time. If not provided, it will fallback to the default limiter set on Application level. If no limiter is set on Application level, it will fallback to the default limiter set by AnyIO.

 None
Example 
from aioclock import Group, AioClock, Forever

email_group = Group()

# consider this as different file
@email_group.task(trigger=Forever())
async def send_email():
    ...

# app.py
aio_clock = AioClock()
aio_clock.include_group(email_group)
Source code in aioclock/group.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    *,
    limiter: Optional[anyio.CapacityLimiter] = None,
):
    """
    Group of tasks that will be run together.

    Best use case is to have a good modularity and separation of concerns.
    For example, you can have a group of tasks that are responsible for sending emails.
    And another group of tasks that are responsible for sending notifications.

    Params:
        limiter:
            Anyio CapacityLimiter. capacity limiter to use to limit the total amount of threads running
            Limiter that will be used to limit the number of tasks that are running at the same time.
            If not provided, it will fallback to the default limiter set on Application level.
            If no limiter is set on Application level, it will fallback to the default limiter set by AnyIO.

    Example:
        ```python

        from aioclock import Group, AioClock, Forever

        email_group = Group()

        # consider this as different file
        @email_group.task(trigger=Forever())
        async def send_email():
            ...

        # app.py
        aio_clock = AioClock()
        aio_clock.include_group(email_group)
        ```

    """
    self._tasks: list[Task] = []
    self._limiter = limiter

task 

task(*, trigger: BaseTrigger)

Decorator to add a task to the AioClock instance. If decorated function is sync, aioclock will run it in a thread pool executor, using AnyIO. But if you try to run the decorated function, it will run in the same thread, blocking the event loop. It is intended to not change all your sync functions to coroutine functions, and they can be used outside of aioclock, if needed.

Parameters:

Name Type Description Default
trigger BaseTrigger

BaseTrigger Trigger that will trigger the task to be running.

 required
Example 
from aioclock import AioClock, Once

app = AioClock()

@app.task(trigger=Once())
async def main():
    print("Hello World")
Source code in aioclock/group.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def task(self, *, trigger: BaseTrigger):
    """
    Decorator to add a task to the AioClock instance.
    If decorated function is sync, aioclock will run it in a thread pool executor, using AnyIO.
    But if you try to run the decorated function, it will run in the same thread, blocking the event loop.
    It is intended to not change all your `sync functions` to coroutine functions,
        and they can be used outside of aioclock, if needed.

    params:
        trigger: BaseTrigger
            Trigger that will trigger the task to be running.

    Example:
        ```python

        from aioclock import AioClock, Once

        app = AioClock()

        @app.task(trigger=Once())
        async def main():
            print("Hello World")
        ```
    """

    def decorator(func):
        @wraps(func)
        async def wrapped_funciton(*args, **kwargs):
            if asyncio.iscoroutinefunction(func):
                return await func(*args, **kwargs)
            else:  # run in threadpool to make sure it's not blocking the event loop
                return await asyncify(func, limiter=self._limiter)(*args, **kwargs)

        self._tasks.append(
            Task(
                func=inject(wrapped_funciton, dependency_overrides_provider=get_provider()),
                trigger=trigger,
            )
        )

        if asyncio.iscoroutinefunction(func):
            return wrapped_funciton

        else:

            @wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)

            return wrapper

    return decorator