Add api

Author: lsbardel

Makefile

@@ -43,3 +43,9 @@ publish:		## release to pypi and github tag
 
 outdated:		## Show outdated packages
 	poetry show -o -a
+
+
+
+.PHONY: example
+example:		## run task scheduler example
+	@poetry run python -m examples.all_features

examples/__init__.py

@@ -0,0 +1,17 @@
+import uvicorn
+from fluid.scheduler import TaskScheduler
+from fluid.tasks_api import task_manager_routes
+from fastapi import FastAPI
+from . import tasks
+
+
+def main():
+    task_manager = TaskScheduler()
+    task_manager.register_from_module(tasks)
+    app = FastAPI()
+    task_manager_routes(app, task_manager)
+    uvicorn.run(app)
+
+
+if __name__ == "__main__":
+    main()

examples/all_features.py

@@ -0,0 +1,37 @@
+import asyncio
+import os
+from datetime import timedelta
+from typing import cast
+
+from fluid.scheduler import TaskRun, every, task
+
+
+@task
+async def dummy(context: TaskRun) -> float:
+    sleep = cast(float, context.params.get("sleep", 0.1))
+    await asyncio.sleep(sleep)
+    if context.params.get("error"):
+        raise RuntimeError("just an error")
+    return sleep
+
+
+@task(schedule=every(timedelta(seconds=1)))
+async def scheduled(context: TaskRun) -> str:
+    """A simple scheduled task"""
+    await asyncio.sleep(0.1)
+    return "OK"
+
+
+@task
+async def disabled(context: TaskRun) -> float:
+    sleep = cast(float, context.params.get("sleep", 0.1))
+    await asyncio.sleep(sleep)
+    return sleep
+
+
+@task(cpu_bound=True)
+async def cpu_bound(context: TaskRun) -> int:
+    await asyncio.sleep(0.1)
+    redis = context.task_manager.broker.redis_cli
+    await redis.setex(context.run_id, os.getpid(), 10)
+    return 0

examples/tasks.py

@@ -0,0 +1,22 @@
+from fastapi import FastAPI
+
+from fluid.scheduler import TaskManager
+from fluid.utils.worker import Workers
+
+from .endpoints import router
+from .service import FastapiAppWorkers
+
+
+def app_workers(app: FastAPI) -> Workers:
+    if workers := getattr(app.state, "workers", None):
+        return workers
+    else:
+        workers = FastapiAppWorkers.setup(app)
+        app.state.workers = workers
+        return workers
+
+
+def task_manager_routes(app: FastAPI, task_manager: TaskManager) -> None:
+    app.state.task_manager = task_manager
+    app.include_router(router, tags=["Tasks"])
+    app_workers(app).add_workers(task_manager)

fluid/tasks_api/__init__.py

@@ -0,0 +1,69 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Path, Request
+from pydantic import BaseModel, Field
+
+from fluid.scheduler import QueuedTask, TaskInfo, TaskManager, TaskPriority
+from fluid.scheduler.errors import UnknownTaskError
+
+router = APIRouter()
+
+
+def get_task_manger(request: Request) -> TaskManager:
+    return request.app.state.task_manager
+
+
+TaskManagerDep = Annotated[TaskManager, Depends(get_task_manger)]
+
+
+class TaskUpdate(BaseModel):
+    enabled: bool = Field(description="Task enabled or disabled")
+
+
+class TaskCreate(BaseModel):
+    name: str = Field(description="Task name")
+    params: dict = Field(description="Task parameters", default_factory=dict)
+    priority: TaskPriority | None = Field(default=None, description="Task priority")
+
+
+@router.get(
+    "/tasks",
+    response_model=list[TaskInfo],
+    summary="List Tasks",
+    description="Retrieve a list of tasks runs",
+)
+async def get_tasks(task_manager: TaskManagerDep) -> list[TaskInfo]:
+    return await task_manager.broker.get_tasks_info()
+
+
+@router.post(
+    "/tasks",
+    response_model=QueuedTask,
+    summary="Queue a new Tasks",
+    description="Queue a new task to be run",
+)
+async def queue_task(
+    task_manager: TaskManagerDep,
+    task: TaskCreate,
+) -> QueuedTask:
+    try:
+        return task_manager.queue(task.name, task.priority, **task.params)
+    except UnknownTaskError as exc:
+        raise HTTPException(status_code=404, detail="Task not found") from exc
+
+
+@router.patch(
+    "/tasks/{task_name}",
+    response_model=TaskInfo,
+    summary="Update a task",
+    description="Update a task configuration and enable/disable it",
+)
+async def patch_task(
+    task_manager: TaskManagerDep,
+    task_update: TaskUpdate,
+    task_name: str = Path(title="Task name"),
+) -> TaskInfo:
+    try:
+        return await task_manager.broker.enable_task(task_name, task_update.enabled)
+    except UnknownTaskError as exc:
+        raise HTTPException(status_code=404, detail="Task not found") from exc

fluid/tasks_api/endpoints.py

@@ -0,0 +1,33 @@
+import os
+import signal
+from typing import Any, Self
+
+from fastapi import FastAPI
+
+from fluid import settings
+from fluid.utils.worker import Worker, Workers
+
+logger = settings.get_logger(__name__)
+
+
+class FastapiAppWorkers(Workers):
+    """An aiohttp runner"""
+
+    @classmethod
+    def setup(cls, app: FastAPI, **kwargs: Any) -> Self:
+        """Setup the app runner"""
+        workers = cls(**kwargs)
+        app.state.workers = workers
+        app.add_event_handler("startup", workers.startup)
+        app.add_event_handler("shutdown", workers.shutdown)
+        return workers
+
+    def bail_out(self, reason: str, code: int = 1) -> None:
+        logger.warning("shutting down due to %s", reason)
+        os.kill(os.getpid(), signal.SIGTERM)
+
+    def get_active_worker(self, *, worker_name: str) -> Worker | None:
+        worker = self._workers.get_worker_by_name(worker_name)
+        if worker and not worker.stopping:
+            return worker
+        return None

fluid/tasks_api/service.py

@@ -18,10 +18,11 @@ async-timeout = "^4.0.3"
 inflection = "^0.5.1"
 redis = "^5.0.1"
 pydantic = "^2.0.3"
+yarl = "^1.9.4"
 aioconsole = {version = "^0.7.0", optional = true }
 sqlalchemy = {version = "^2.0.23", optional = true }
 fastapi = {version = "^0.110.0", optional = true}
-yarl = "^1.9.4"
+uvicorn = {version = "^0.29.0", optional = true}
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.1.1"
@@ -41,7 +42,7 @@ optional = true
 mkdocs-material = "^9.5.15"
 
 [tool.poetry.extras]
-fastapi = ["fastapi", "aioconsole"]
+fastapi = ["fastapi", "aioconsole", "uvicorn"]
 db = ["sqlalchemy"]
 
 [build-system]