Added scheduler. (#47)

* Initial scheduler implementation.

Signed-off-by: Pavel Kirilin <[email protected]>

Author: s3rius

docs/available-components/README.md

@@ -0,0 +1,13 @@
+---
+order: 1
+dir:
+    order: 3
+---
+
+# Available components
+
+In this section, you can find a list of officially supported plugins for the taskiq.
+
+* [Available brokers](./brokers.md)
+* [Available result backends](./result-backends.md)
+* [Available schedule sources](./scheduler-sources.md)

docs/guide/available-brokers.md renamed to docs/available-components/brokers.md

@@ -1,5 +1,5 @@
 ---
-order: 5
+order: 2
 ---
 
 # Available brokers

docs/guide/result_backends.md renamed to docs/available-components/result-backends.md

@@ -1,8 +1,8 @@
 ---
-order: 6
+order: 3
 ---
 
-# Result backends
+# Available result backends
 
 Result backends are used to store execution results.
 This includes:

docs/available-components/schedule-sources.md

@@ -0,0 +1,59 @@
+---
+order: 4
+---
+
+
+# Available schedule sources
+
+These objects are used to fetch current schedule for tasks.
+Currently we have only one schedule source.
+
+
+## LabelScheduleSource
+
+This source parses labels of tasks, and if it finds a `schedule` label, it considers this task as scheduled.
+
+The format of the schedule label is the following:
+
+```python
+@broker.task(
+    schedule=[
+        {
+            "cron": "* * * * *", # type: str, required argument.
+            "args": [], # type List[Any] | None, can be omitted.
+            "kwargs": {}, # type: Dict[str, Any] | None, can be omitted.
+            "labels": {}, # type: Dict[str, Any] | None, can be omitted.
+        }
+    ]
+)
+async def my_task():
+    ...
+```
+
+Parameters:
+* `cron` - crontab string when to run the task.
+* `args` - args to use, when invoking the task.
+* `kwargs` - key-word arguments to use when invoking the task.
+* `labels` - additional labels to use wehn invoking the task.
+
+Usage:
+
+```python
+from taskiq.scheduler import TaskiqScheduler
+from taskiq.schedule_sources import LabelScheduleSource
+
+broker = ...
+
+scheduler = TaskiqScheduler(
+    broker=broker,
+    sources=[LabelScheduleSource(broker)],
+)
+```
+
+
+::: warning Cool notice!
+
+In order to resolve all labels correctly, don't forget to import
+all task modules using CLI interface.
+
+:::

docs/examples/extending/result_backend.py

@@ -0,0 +1,44 @@
+from typing import TypeVar
+
+from taskiq import TaskiqResult
+from taskiq.abc.result_backend import AsyncResultBackend
+
+_ReturnType = TypeVar("_ReturnType")
+
+
+class MyResultBackend(AsyncResultBackend[_ReturnType]):
+    async def startup(self) -> None:
+        """Do something when starting broker."""
+
+    async def shutdown(self) -> None:
+        """Do something on shutdown."""
+
+    async def set_result(
+        self,
+        task_id: str,
+        result: TaskiqResult[_ReturnType],
+    ) -> None:
+        # Here you must set result somewhere.
+        pass
+
+    async def get_result(
+        self,
+        task_id: str,
+        with_logs: bool = False,
+    ) -> TaskiqResult[_ReturnType]:
+        # Here you must retrieve result by id.
+
+        # Logs is a part of a result.
+        # Here we have a parameter whether you want to
+        # fetch result with logs or not, because logs
+        # can have a lot of info and sometimes it's critical
+        # to get only needed information.
+        pass
+
+    async def is_result_ready(
+        self,
+        task_id: str,
+    ) -> bool:
+        # This function checks if result of a task exists,
+        # without actual fetching the result.
+        pass

docs/examples/extending/schedule_source.py

@@ -0,0 +1,28 @@
+from typing import List
+
+from taskiq import ScheduledTask, ScheduleSource
+
+
+class MyScheduleSource(ScheduleSource):
+    async def startup(self) -> None:
+        """Do something when starting broker."""
+
+    async def shutdown(self) -> None:
+        """Do something on shutdown."""
+
+    async def get_schedules(self) -> List["ScheduledTask"]:
+        # Here you must return list of scheduled tasks from your source.
+        return [
+            ScheduledTask(
+                task_name="",
+                labels={},
+                args=[],
+                kwargs={},
+                cron="* * * * *",
+            ),
+        ]
+
+    # This method is optional. You may not implement this.
+    # It's just a helper to people to be able to interact with your source.
+    async def add_schedule(self, schedule: "ScheduledTask") -> None:
+        return await super().add_schedule(schedule)

docs/examples/schedule/intro.py

@@ -0,0 +1,16 @@
+from taskiq_aio_pika import AioPikaBroker
+
+from taskiq.schedule_sources import LabelScheduleSource
+from taskiq.scheduler import TaskiqScheduler
+
+broker = AioPikaBroker("amqp://guest:guest@localhost:5672/")
+
+scheduler = TaskiqScheduler(
+    broker=broker,
+    sources=[LabelScheduleSource(broker)],
+)
+
+
+@broker.task(schedule=[{"cron": "*/5 * * * *", "args": [1]}])
+async def heavy_task(value: int) -> int:
+    return value + 1

docs/examples/schedule/without_schedule.py

@@ -0,0 +1,8 @@
+from taskiq_aio_pika import AioPikaBroker
+
+broker = AioPikaBroker("amqp://guest:guest@localhost:5672/")
+
+
+@broker.task
+async def heavy_task(value: int) -> int:
+    return value + 1

docs/extending-taskiq/README.md

@@ -10,3 +10,12 @@ Taskiq is super extendable. The core library comes with different abstract class
 You can implement these abstract classes to extend functionality.
 
 All abstract classes can be found in `taskiq.abc` package.
+
+
+## Contents:
+
+* [Brokers](./broker.md)
+* [Brokers](./broker.md)
+* [Result backends](./resutl-backend.md)
+* [CLI](./cli.md)
+* [Schedule sources](./schedule-sources.md)

docs/extending-taskiq/cli.md

@@ -27,7 +27,7 @@ from setuptools import setup
 setup(
     # ...,
     entry_points={
-        'taskiq-cli': [
+        'taskiq_cli': [
             'demo = my_project.cmd:MyCommand',
         ]
     }
@@ -37,14 +37,14 @@ setup(
 @tab setuptools pyproject.toml
 
 ```toml
-[project.entry-points.taskiq-cli]
+[project.entry-points.taskiq_cli]
 demo = "my_project.cmd:MyCommand"
 ```
 
 @tab poetry
 
 ```toml
-[tool.poetry.plugins.taskiq-cli]
+[tool.poetry.plugins.taskiq_cli]
 demo = "my_project.cmd:MyCommand"
 ```