Powerful exec command (#53)

* More docs

* command line params

* exec command

Author: lsbardel

.vscode/launch.json

@@ -33,5 +33,22 @@
                 "RedirectOutput"
             ]
         },
+        {
+            "name": "AIO-FLUID EXAMPLE exec",
+            "type": "python",
+            "request": "launch",
+            "program": "${workspaceFolder}/examples",
+            "cwd": "${workspaceFolder}",
+            "justMyCode": false,
+            "args": [
+                "exec",
+                "add",
+                "--a",
+                "1",
+            ],
+            "debugOptions": [
+                "RedirectOutput"
+            ]
+        },
     ]
 }

docs/reference/task_cli.md

@@ -0,0 +1,20 @@
+# Task Manager Cli
+
+Command line tools for task manager applications.
+
+This modules requires the `cli` extra to be installed.
+
+```bash
+$ pip install aio-fluid[cli]
+```
+It can be imported from `fluid.scheduler.cli`:
+
+```python
+from fastapi.scheduler.cli import TaskManagerCLI
+
+if __name__ == "__main__":
+    cli = TaskManagerCLI("path.to:task_app")
+    cli()
+```
+
+::: fluid.scheduler.cli.TaskManagerCLI

docs/reference/task_manager.md

@@ -8,7 +8,6 @@ from fastapi.scheduler import TaskManager
 
 ::: fluid.scheduler.TaskManager
 
-
 ::: fluid.scheduler.TaskManagerConfig
 
 ::: fluid.scheduler.consumer.TaskDispatcher

docs/reference/task_scheduler.md

@@ -0,0 +1,13 @@
+# Task Scheduler
+
+The task scheduler [TaskScheduler][fluid.scheduler.TaskScheduler] inherits
+from the [TaskConsumer][fluid.scheduler.TaskConsumer] to add scheduling of
+periodic tasks.
+
+It can be imported from `fluid.scheduler`:
+
+```python
+from fastapi.scheduler import TaskScheduler
+```
+
+::: fluid.scheduler.TaskScheduler

docs/reference/utils.md

@@ -0,0 +1,3 @@
+# Utils
+
+::: fluid.utils.lazy.LazyGroup

docs/tutorials/task_app.md

@@ -0,0 +1,87 @@
+# Task Queue App
+
+The `fluid.scheduler` module is a simple yet powerful distributed task producer ([TaskScheduler][fluid.scheduler.TaskScheduler]) and consumer ([TaskConsumer][fluid.scheduler.TaskConsumer]) system for executing tasks.
+The middleware for distributing tasks can be configured via the [TaskBroker][fluid.scheduler.TaskBroker] interface.
+
+A redis task broker is provided for convenience.
+
+## Tasks Consumer
+
+Create a task consumer, register tasks from modules, and run the consumer.
+
+```python
+import asyncio
+from typing import Any
+from fluid.scheduler import TaskConsumer
+import task_module_a, task_module_b
+
+
+def task_consumer(**kwargs: Any) -> TaskConsumer:
+    consumer = TaskConsumer(**kwargs)
+    consumer.register_from_module(task_module_a)
+    consumer.register_from_module(task_module_b)
+    return consumer
+
+
+if __name__ == "__main__":
+    consumer = task_consumer()
+    asyncio.run(consumer.run())
+```
+
+## FastAPI Integration
+
+The `TaskConsumer` can be integrated with FastAPI so that
+tasks can be queued via HTTP requests.
+
+```python
+import uvicorn
+from fluid.scheduler.endpoints import setup_fastapi
+
+if __name__ == "__main__":
+    consumer = task_consumer()
+    app = setup_fastapi(consumer)
+    uvicorn.run(app)
+```
+
+You can test via the example provided
+
+```bash
+$ python -m examples.simple_fastapi
+```
+
+and check the openapi UI at [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs).
+
+
+## Task App Command Line
+
+The [TaskConsumer][fluid.scheduler.TaskConsumer] or [TaskScheduler][fluid.scheduler.TaskScheduler] can be run with the command line tool to allow for an even richer API.
+
+```python
+from fluid.scheduler.cli import TaskManagerCLI
+
+if __name__ == "__main__":
+    consumer = task_consumer()
+    TaskManagerCLI(setup_fastapi(consumer))()
+```
+
+This features requires to install the package with the `cli` extra.
+
+```bash
+$ pip install aio-fluid[cli]
+```
+
+```bash
+$ python -m examples.simple_cli
+Usage: python -m examples.simple_cli [OPTIONS] COMMAND [ARGS]...
+
+Options:
+  --help  Show this message and exit.
+
+Commands:
+  exec   Execute a registered task
+  ls     List all tasks with their schedules
+  serve  Start app server
+```
+
+The command line tools provides a powerful interface to execute tasks, parameters are
+passed as optional arguments using he standard click interface.

docs/tutorials/task_queue.md

@@ -1,10 +1,4 @@
-# Task Queue
-
-The `fluid.scheduler` module is a simple yet powerful distributed task producer (TaskScheduler) and consumer (TaskConsumer) system for executing tasks.
-The middleware for distributing tasks can be configured via the Broker interface.
-A redis broker is provided for convenience.
-
-## Tasks
+# Tasks
 
 Tasks are standard python async functions decorated with the [@task][fluid.scheduler.task] decorator.
 
@@ -42,7 +36,7 @@ async def say_hi(ctx: TaskRun[TaskParams]) -> None:
 
 ## Task Types
 
-There are two types of tasks implemented
+There are few types of tasks implemented, lets take a look at them.
 
 ### IO Bound Tasks
 
@@ -89,10 +83,22 @@ Both IO and CPU bound tasks can be periodically scheduled via the `schedule` key
 There are two types of scheduling, the most common is the [every][fluid.scheduler.every] function that takes a `timedelta` object.
 
 ```python
+import asyncio
 from datetime import timedelta
-from fluid.scheduler import task, TaskContext, every
+from fluid.scheduler import task, TaskRun, every
 
 @task(schedule=every(timedelta(seconds=1)))
 async def scheduled(ctx: TaskRun) -> None:
     await asyncio.sleep(0.1)
 ```
+
+You can also use the [crontab][fluid.scheduler.crontab] function to schedule tasks using cron expressions.
+
+```python
+import asyncio
+from fluid.scheduler import task, TaskRun, crontab
+
+@task(schedule=crontab(hours='*/2'))
+async def scheduled(ctx: TaskRun) -> None:
+    await asyncio.sleep(0.1)
+```

examples/simple.py

@@ -0,0 +1,8 @@
+import asyncio
+
+from examples.tasks import task_scheduler
+from fluid.utils import log
+
+if __name__ == "__main__":
+    log.config()
+    asyncio.run(task_scheduler().run())

examples/simple_cli.py

@@ -0,0 +1,12 @@
+import dotenv
+
+dotenv.load_dotenv()
+
+from fluid.scheduler.cli import TaskManagerCLI  # isort:skip    # noqa: E402
+
+
+if __name__ == "__main__":
+    from examples.tasks import task_app
+
+    task_manager_cli = TaskManagerCLI(task_app())
+    task_manager_cli()

examples/simple_fastapi.py

@@ -0,0 +1,10 @@
+import uvicorn
+
+from examples.tasks import task_scheduler
+from fluid.scheduler.endpoints import setup_fastapi
+from fluid.utils import log
+
+if __name__ == "__main__":
+    log.config()
+    app = setup_fastapi(task_scheduler())
+    uvicorn.run(app)

examples/tasks/__init__.py

@@ -12,10 +12,14 @@
 from fluid.scheduler.endpoints import setup_fastapi
 
 
-def task_app() -> FastAPI:
+def task_scheduler() -> TaskScheduler:
     task_manager = TaskScheduler()
     task_manager.register_from_dict(globals())
-    return setup_fastapi(task_manager)
+    return task_manager
+
+
+def task_app() -> FastAPI:
+    return setup_fastapi(task_scheduler())
 
 
 class Sleep(BaseModel):
@@ -38,8 +42,8 @@ async def scheduled(context: TaskRun) -> None:
 
 
 class AddValues(BaseModel):
-    a: float = 0
-    b: float = 0
+    a: float = Field(default=0, description="First number to add")
+    b: float = Field(default=0, description="Second number to add")
 
 
 @task

fluid/scheduler/cli.py

@@ -5,11 +5,13 @@
 import click
 import uvicorn
 from fastapi import FastAPI
+from pydanclick import from_pydantic
+from pydantic import BaseModel
 from rich.console import Console
 from rich.table import Table
 from uvicorn.importer import import_from_string
 
-from fluid.utils import log
+from fluid.utils import log as log_
 from fluid.utils.lazy import LazyGroup
 
 if TYPE_CHECKING:
@@ -21,6 +23,13 @@
 
 
 class TaskManagerCLI(LazyGroup):
+    """CLI for TaskManager
+
+    This class provides a CLI for a TaskManager Application.
+
+    It requires to install the `cli` extra dependencies.
+    """
+
     def __init__(
         self,
         task_manager_app: TaskManagerApp,
@@ -49,10 +58,38 @@ def ctx_task_manager(ctx: click.Context) -> TaskManager:
     return ctx_app(ctx).state.task_manager
 
 
+class ExecuteTasks(click.Group):
+    def list_commands(self, ctx: click.Context) -> list[str]:
+        task_manager = ctx_task_manager(ctx)
+        return sorted(task_manager.registry)
+
+    def get_command(self, ctx: click.Context, cmd_name: str) -> click.Command | None:
+        task_manager = ctx_task_manager(ctx)
+        task = task_manager.registry.get(cmd_name)
+        if task is None:
+            raise click.ClickException(f"Task {cmd_name} not found")
+
+        @click.command(cmd_name, help=task.short_description)
+        @click.option("--log", is_flag=True, help="Show logs")
+        @from_pydantic(task.params_model)
+        def execute_task(log: bool, **kwargs: Any) -> None:
+            if log:
+                log_.config()
+            params = {}
+            for value in kwargs.values():
+                if isinstance(value, BaseModel):
+                    params.update(value.model_dump())
+            run = task_manager.execute_sync(cmd_name, **params)
+            console = Console()
+            console.print(task_run_table(run))
+
+        return execute_task
+
+
 @click.command()
 @click.pass_context
 def ls(ctx: click.Context) -> None:
-    """list all tasks"""
+    """List all tasks with their schedules"""
     task_manager = ctx_task_manager(ctx)
     table = Table(title="Tasks")
     table.add_column("Name", style="cyan", no_wrap=True)
@@ -75,23 +112,6 @@ def ls(ctx: click.Context) -> None:
     console.print(table)
 
 
-@click.command()
-@click.pass_context
-@click.argument("task")
-@click.option(
-    "--dry-run",
-    is_flag=True,
-    help="dry run (if the tasks supports it)",
-    default=False,
-)
-def execute(ctx: click.Context, task: str, dry_run: bool) -> None:
-    """execute a task"""
-    task_manager = ctx_task_manager(ctx)
-    run = task_manager.execute_sync(task, dry_run=dry_run)
-    console = Console()
-    console.print(task_run_table(run))
-
-
 @click.command("serve", short_help="Start app server.")
 @click.option(
     "--host",
@@ -116,18 +136,20 @@ def execute(ctx: click.Context, task: str, dry_run: bool) -> None:
 )
 @click.pass_context
 def serve(ctx: click.Context, host: str, port: int, reload: bool) -> None:
-    """Run the service."""
+    """Run the service"""
     task_manager_app = ctx_task_manager_app(ctx)
     uvicorn.run(
         task_manager_app,
         port=port,
         host=host,
         log_level="info",
         reload=reload,
-        log_config=log.config(),
+        log_config=log_.config(),
     )
 
 
+execute = ExecuteTasks(name="exec", help="Execute a registered task")
+
 DEFAULT_COMMANDS = (ls, execute, serve)
 
 

fluid/utils/lazy.py

@@ -5,6 +5,17 @@
 
 
 class LazyGroup(click.Group):
+    """A click Group that can lazily load subcommands
+
+    This class extends the click.Group class to allow for subcommands to be
+    lazily loaded from a module path.
+
+    It is useful when you have a large number of subcommands that you don't
+    want to load until they are actually needed.
+
+    Available with the `cli` extra dependencies.
+    """
+
     def __init__(
         self,
         *,