Plugins (FastAPI included)

Extensions for aioclock.

AioClock is very extensible, and you can add your own extensions to it. The extension would allow you to interact with your AioClock instance, from different layers of your application. For instance, the FastAPI plugin allows you to run a specific task immediately from an HTTP API, or see your tasks in an HTTP API, and when they are going to run next.

FastAPI extension to manage the tasks of the AioClock instance in HTTP Layer.

Use cases

• Expose the tasks of the AioClock instance in an HTTP API.
• Show to your client which task is going to be run next, and at which time.
• Run a specific task from an HTTP API immidiately if needed.

To use FastAPI Extension, please make sure you do pip install aioclock[fastapi].

make_fastapi_router

make_fastapi_router(
    aioclock: AioClock,
    router: Union[APIRouter, None] = None,
)

Make a FastAPI router that exposes the tasks of the AioClock instance and its external python API in HTTP Layer. You can pass a router to this function, and have dependencies injected in the router, or any authorization logic that you want to have.

Parameters:

Name	Type	Description	Default
aioclock	AioClock	AioClock instance to get the tasks from.	required
router	Union[APIRouter, None]	FastAPI router to add the routes to. If not provided, a new router will be created.	None

Example

import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI

from aioclock import AioClock
from aioclock.ext.fast import make_fastapi_router
from aioclock.triggers import Every, OnStartUp

clock_app = AioClock()

@clock_app.task(trigger=OnStartUp())
async def startup():
    print("Starting...")

@clock_app.task(trigger=Every(seconds=3600))
async def foo():
    print("Foo is processing...")


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(clock_app.serve())
    yield

    try:
        task.cancel()
        await task
    except asyncio.CancelledError:
        ...


app = FastAPI(lifespan=lifespan)
app.include_router(make_fastapi_router(clock_app))

if __name__ == "__main__":
    import uvicorn
    # uvicorn.run(app)

Source code in aioclock/ext/fast.py

def make_fastapi_router(aioclock: AioClock, router: Union[APIRouter, None] = None):
    """Make a FastAPI router that exposes the tasks of the AioClock instance and its external python API in HTTP Layer.
    You can pass a router to this function, and have dependencies injected in the router, or any authorization logic that you want to have.


    params:
        aioclock: AioClock instance to get the tasks from.
        router: FastAPI router to add the routes to. If not provided, a new router will be created.

    Example:
        ```python
        import asyncio
        from contextlib import asynccontextmanager

        from fastapi import FastAPI

        from aioclock import AioClock
        from aioclock.ext.fast import make_fastapi_router
        from aioclock.triggers import Every, OnStartUp

        clock_app = AioClock()

        @clock_app.task(trigger=OnStartUp())
        async def startup():
            print("Starting...")

        @clock_app.task(trigger=Every(seconds=3600))
        async def foo():
            print("Foo is processing...")


        @asynccontextmanager
        async def lifespan(app: FastAPI):
            task = asyncio.create_task(clock_app.serve())
            yield

            try:
                task.cancel()
                await task
            except asyncio.CancelledError:
                ...


        app = FastAPI(lifespan=lifespan)
        app.include_router(make_fastapi_router(clock_app))

        if __name__ == "__main__":
            import uvicorn
            # uvicorn.run(app)
        ```
    """
    router = router or APIRouter()

    @router.get("/tasks")
    async def get_tasks() -> list[TaskMetadata]:
        return await get_metadata_of_all_tasks(aioclock)

    @router.post("/task/{task_id}")
    async def run_task(task_id: UUID):
        try:
            await run_specific_task(task_id, aioclock)
        except TaskIdNotFound:
            raise HTTPException(status_code=404, detail="Task not found")

    return router