External API

External API of the aioclock package, that can be used to interact with the AioClock instance. This module could be very useful if you intend to use aioclock in a web application or a CLI tool.

Other tools and extension are written from this tool.

Note when writing to aioclock API and changing its state.

Right now the state of AioClock instance is on the memory level, so if you write an API and change a task's trigger time, it will not persist. In future we might store the state of AioClock instance in a database, so that it always remains same. But this is a bit tricky and implicit because then your code gets ignored and database is preferred over the codebase. For now you may consider it as a way to change something without redeploying the application, but it is not very recommended to write.

TaskMetadata

Bases: BaseModel

Metadata of the task that is included in the AioClock instance.

Attributes:

Name	Type	Description
id	UUID	UUID: Task ID that is unique for each task, and changes every time you run the aioclock app. In future we might store task ID in a database, so that it always remains same.
trigger	Union[TriggerT, Any]	Union[TriggerT, Any]: Trigger that is used to run the task, type is also any to ease implementing new triggers.
task_name	str	str: Name of the task function.

run_specific_task async

run_specific_task(task_id: UUID, app: AioClock)

Run a specific task immediately by its ID, from the AioClock instance.

Parameters:

Name	Type	Description	Default
task_id	UUID	Task ID that is unique for each task, and changes every time you run the aioclock app. In future we might store task ID in a database, so that it always remains same.	required
app	AioClock	AioClock instance to run the task from.	required

Example

from aioclock import  AioClock, Once
from aioclock.api import run_specific_task

app = AioClock()

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

async def some_other_func():
    await run_specific_task(app._tasks[0].id, app)

Source code in aioclock/api.py

async def run_specific_task(task_id: UUID, app: AioClock):
    """Run a specific task immediately by its ID, from the AioClock instance.

    params:
        task_id: Task ID that is unique for each task, and changes every time you run the aioclock app.
            In future we might store task ID in a database, so that it always remains same.
        app: AioClock instance to run the task from.

    Example:
        ```python
        from aioclock import  AioClock, Once
        from aioclock.api import run_specific_task

        app = AioClock()

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

        async def some_other_func():
            await run_specific_task(app._tasks[0].id, app)
        ```

    """
    task = next((task for task in app._tasks if task.id == task_id), None)
    if not task:
        raise TaskIdNotFound
    return await run_with_injected_deps(task.func)

run_with_injected_deps async

run_with_injected_deps(
    func: Callable[P, Awaitable[T]]
) -> T

Runs an aioclock decorated function, with all the dependencies injected.

Can be used to run a task function with all the dependencies injected.

Parameters:

Name	Type	Description	Default
func	Callable[P, Awaitable[T]]	Function to run with all the dependencies injected. Must be decorated with @app.task decorator.	required

Example

from aioclock import Once, AioClock, Depends
from aioclock.api import run_with_injected_deps

app = AioClock()

def some_dependency():
    return 1

@app.task(trigger=Once())
async def main(bar: int = Depends(some_dependency)):
    print("Hello World")
    return bar

async def some_other_func():
    foo = await run_with_injected_deps(main)
    assert foo == 1

Source code in aioclock/api.py

async def run_with_injected_deps(func: Callable[P, Awaitable[T]]) -> T:
    """Runs an aioclock decorated function, with all the dependencies injected.

    Can be used to run a task function with all the dependencies injected.

    params:
        func: Function to run with all the dependencies injected. Must be decorated with `@app.task` decorator.

    Example:
        ```python
        from aioclock import Once, AioClock, Depends
        from aioclock.api import run_with_injected_deps

        app = AioClock()

        def some_dependency():
            return 1

        @app.task(trigger=Once())
        async def main(bar: int = Depends(some_dependency)):
            print("Hello World")
            return bar

        async def some_other_func():
            foo = await run_with_injected_deps(main)
            assert foo == 1
        ```

    """
    return await inject(func, dependency_overrides_provider=get_provider())()  # type: ignore

get_metadata_of_all_tasks async

get_metadata_of_all_tasks(
    app: AioClock,
) -> list[TaskMetadata]

Get metadata of all tasks that are included in the AioClock instance.

This function can be used to mutate the TaskMetadata object, i.e to change the trigger of a task. But for now it is yet not recommended to do this, as you might experience some unexpected behavior. But in next versions, I'd like to make it more stable and reliable on mutating the data.

Parameters:

Name	Type	Description	Default
app	AioClock	AioClock instance to get the metadata of all tasks.	required

Example

from aioclock import AioClock, Once
from aioclock.api import get_metadata_of_all_tasks

app = AioClock()
@app.task(trigger=Once())
async def main(): ...

async def some_other_func():
    metadata = await get_metadata_of_all_tasks(app)

Source code in aioclock/api.py

async def get_metadata_of_all_tasks(app: AioClock) -> list[TaskMetadata]:
    """Get metadata of all tasks that are included in the AioClock instance.

    This function can be used to mutate the `TaskMetadata` object, i.e to change the trigger of a task.
    But for now it is yet not recommended to do this, as you might experience some unexpected behavior.
    But in next versions, I'd like to make it more stable and reliable on mutating the data.

    params:
        app: AioClock instance to get the metadata of all tasks.

    Example:
        ```python
        from aioclock import AioClock, Once
        from aioclock.api import get_metadata_of_all_tasks

        app = AioClock()
        @app.task(trigger=Once())
        async def main(): ...

        async def some_other_func():
            metadata = await get_metadata_of_all_tasks(app)
        ```
    """
    return [
        TaskMetadata(
            id=task.id,
            trigger=task.trigger,
            task_name=task.func.__name__,
        )
        for task in app._get_tasks(exclude_type=set())
    ]