Added new serializers. (#217)

Author: s3rius

.github/workflows/test.yml

@@ -21,7 +21,7 @@ jobs:
           python-version: "3.11"
           cache: "poetry"
       - name: Install deps
-        run: poetry install
+        run: poetry install --all-extras
       - name: Run lint check
         run: poetry run pre-commit run -a ${{ matrix.cmd }}
   pytest:
@@ -44,7 +44,7 @@ jobs:
           python-version: "${{ matrix.py_version }}"
           cache: "poetry"
       - name: Install deps
-        run: poetry install
+        run: poetry install --all-extras
       - name: Run pytest check
         run: poetry run pytest -vv -n auto --cov="taskiq" .
       - name: Generate report

docs/guide/message-format.md

@@ -0,0 +1,113 @@
+---
+order: 11
+---
+
+# Taskiq message format
+
+Taskiq doesn't force you to use any specific message format. We define default message format,
+but you can use any format you want.
+
+The default message format is:
+
+
+::: tabs
+
+@tab example
+
+```json
+{
+    "task_name": "my_project.module1.task",
+    "args": [1, 2, 3],
+    "kwargs": {"a": 1, "b": 2, "c": 3},
+    "labels": {
+        "label1": "value1",
+        "label2": "value2"
+    }
+}
+```
+
+@tab json schema
+
+```json
+{
+  "properties": {
+    "task_id": {
+      "title": "Task Id",
+      "type": "string"
+    },
+    "task_name": {
+      "title": "Name of the task",
+      "type": "string"
+    },
+    "labels": {
+      "title": "Additional labels",
+      "type": "object"
+    },
+    "args": {
+      "items": {},
+      "title": "Arguments",
+      "type": "array"
+    },
+    "kwargs": {
+      "title": "Keyword arguments",
+      "type": "object"
+    }
+  },
+  "required": [
+    "task_id",
+    "task_name",
+    "labels",
+    "args",
+    "kwargs"
+  ],
+  "type": "object"
+}
+```
+
+:::
+
+But this can be easily changed by creating your own implementation of the TaskiqFormatter class or TaskiqSerializer class.
+
+
+### Serializers
+
+Serializers define the format of the message but not the structure. For example, if you want to use msgpack or ORJson to serialize your message, you should update the serializer of your broker.
+
+Be default, Taskiq uses JSON serializer. But we also have some implementations of other serializers:
+
+* ORJSONSerializer - faster [JSON implementation](https://pypi.org/project/orjson/). Also, it supports datetime and UUID serialization.
+* MSGPackSerializer - [MsgPack](https://pypi.org/project/msgpack/) format serializer. It might be useful to send less data over the network.
+* CBORSerializer - [CBOR](https://pypi.org/project/cbor2/) format serializer. It is also has a smaller size than JSON.
+
+To define your own serializer, you have to subclass the TaskiqSerializer class and implement `dumpb` and `loadb` methods. You can take a look at the existing implementations from the `taskiq.serializers` module.
+
+To install taskiq with libraries for non-JSON serializers, you should install taskiq with extras.
+
+::: tabs
+
+@tab orjson
+
+```bash
+pip install "taskiq[orjson]"
+```
+
+@tab msgpack
+
+```bash
+pip install "taskiq[msgpack]"
+```
+
+@tab cbor
+
+```bash
+pip install "taskiq[cbor]"
+```
+
+:::
+
+### Formatters
+
+Formatters define the format of the message. It might be useful if you'd like to send a task to a celery worker for a different project. You can do it in seriazier as well, but formatters give you correct type hints.
+
+By default we use a formatter that dumps the message to dict and serializes it using serializer. But you can define your own formatter to send a message in any format you want. To define a new formatter, you have to subclass the TaskiqFormatter class and implement `dumps` and `loads` methods.
+As an example, you can take a look at the `JSONFormatter` from `taskiq.formatters` implementation.

poetry.lock

@@ -39,11 +39,14 @@ prometheus_client = { version = "^0", optional = true }
 # For ZMQBroker
 pyzmq = { version = "^23.2.0", optional = true }
 # For speed
-uvloop = { version = ">=0.16.0,<1", optional = true }
+uvloop = { version = ">=0.16.0,<1", optional = true, markers = "sys_platform != 'win32'" }
 # For hot-reload.
 watchdog = { version = "^2.1.9", optional = true }
 gitignore-parser = { version = "^0", optional = true }
 pytz = "*"
+orjson = { version = "^3.9.9", optional = true }
+msgpack = { version = "^1.0.7", optional = true }
+cbor2 = { version = "^5.4.6", optional = true }
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.1.2"
@@ -68,6 +71,9 @@ zmq = ["pyzmq"]
 uv = ["uvloop"]
 metrics = ["prometheus_client"]
 reload = ["watchdog", "gitignore-parser"]
+orjson = ["orjson"]
+msgpack = ["msgpack"]
+cbor = ["cbor2"]
 
 [tool.poetry.scripts]
 taskiq = "taskiq.__main__:main"
@@ -160,12 +166,12 @@ line-length = 88
 
 [tool.ruff.per-file-ignores]
 "tests/*" = [
-    "S101", # Use of assert detected
-    "S301", # Use of pickle detected
-    "D103", # Missing docstring in public function
+    "S101",   # Use of assert detected
+    "S301",   # Use of pickle detected
+    "D103",   # Missing docstring in public function
     "SLF001", # Private member accessed
-    "S311", # Standard pseudo-random generators are not suitable for security/cryptographic purposes
-    "D101", # Missing docstring in public class
+    "S311",   # Standard pseudo-random generators are not suitable for security/cryptographic purposes
+    "D101",   # Missing docstring in public class
 ]
 
 [tool.ruff.pydocstyle]
@@ -176,7 +182,4 @@ ignore-decorators = ["typing.overload"]
 allow-magic-value-types = ["int", "str", "float", "tuple"]
 
 [tool.ruff.flake8-bugbear]
-extend-immutable-calls = [
-    "taskiq_dependencies.Depends",
-    "taskiq.TaskiqDepends",
-]
+extend-immutable-calls = ["taskiq_dependencies.Depends", "taskiq.TaskiqDepends"]

pyproject.toml

@@ -495,6 +495,16 @@ def with_serializer(
         self.serializer = serializer
         return self
 
+    def with_formatter(self, formatter: "TaskiqFormatter") -> "Self":
+        """
+        Set new formatter and return an updated broker.
+
+        :param formatter: new formatter.
+        :return: self
+        """
+        self.formatter = formatter
+        return self
+
     def _register_task(
         self,
         task_name: str,

taskiq/abc/broker.py

@@ -1 +1,12 @@
 """Taskiq serializers."""
+from .cbor_serializer import CBORSerializer
+from .json_serializer import JSONSerializer
+from .msgpack_serializer import MSGPackSerializer
+from .orjson_serializer import ORJSONSerializer
+
+__all__ = [
+    "JSONSerializer",
+    "ORJSONSerializer",
+    "MSGPackSerializer",
+    "CBORSerializer",
+]

taskiq/serializers/__init__.py

@@ -0,0 +1,63 @@
+import datetime
+from typing import Any, Callable, Optional
+
+from taskiq.abc.serializer import TaskiqSerializer
+
+try:
+    import cbor2
+except ImportError:
+    cbor2 = None  # type: ignore
+
+
+class CBORSerializer(TaskiqSerializer):
+    """
+    Taskiq serializer using cbor2 library.
+
+    See https://cbor2.readthedocs.io/en/stable/ for more information.
+    """
+
+    def __init__(
+        self,
+        datetime_as_timestamp: bool = True,
+        timezone: Optional[datetime.tzinfo] = None,
+        value_sharing: bool = False,
+        default: Optional[Callable[[Any, Any], Any]] = None,
+        canonical: bool = False,
+        date_as_datetime: bool = True,
+        string_referencing: bool = True,
+        # Decoder options
+        tag_hook: "Optional[Callable[[cbor2.CborDecoder, Any], Any]]" = None,
+        object_hook: Optional[Callable[[Any], Any]] = None,
+    ) -> None:
+        if cbor2 is None:
+            raise ImportError("cbor2 is not installed")
+        self.datetime_as_timestamp = datetime_as_timestamp
+        self.timezone = timezone
+        self.value_sharing = value_sharing
+        self.default = default
+        self.canonical = canonical
+        self.date_as_datetime = date_as_datetime
+        self.string_referencing = string_referencing
+        self.tag_hook = tag_hook
+        self.object_hook = object_hook
+
+    def dumpb(self, value: Any) -> bytes:
+        """Dump value to bytes."""
+        return cbor2.dumps(
+            value,
+            datetime_as_timestamp=self.datetime_as_timestamp,
+            timezone=self.timezone,
+            value_sharing=self.value_sharing,
+            default=self.default,
+            canonical=self.canonical,
+            date_as_datetime=self.date_as_datetime,
+            string_referencing=self.string_referencing,
+        )
+
+    def loadb(self, value: bytes) -> Any:
+        """Load value from bytes."""
+        return cbor2.loads(
+            value,
+            tag_hook=self.tag_hook,
+            object_hook=self.object_hook,
+        )

taskiq/serializers/cbor_serializer.py

@@ -0,0 +1,40 @@
+from typing import Any, Callable, Optional
+
+from taskiq.abc.serializer import TaskiqSerializer
+
+try:
+    import msgpack
+except ImportError:
+    msgpack = None  # type: ignore
+
+
+class MSGPackSerializer(TaskiqSerializer):
+    """Taskiq serializer using msgpack library."""
+
+    def __init__(
+        self,
+        default: Optional[Callable[[Any], Any]] = None,
+        use_single_float: bool = False,
+        use_bin_type: bool = True,
+        datetime: bool = True,
+    ) -> None:
+        if msgpack is None:
+            raise ImportError("msgpack is not installed")
+        self.default = default
+        self.use_single_float = use_single_float
+        self.use_bin_type = use_bin_type
+        self.datetime = datetime
+
+    def dumpb(self, value: Any) -> bytes:
+        """Dump value to bytes."""
+        return msgpack.packb(
+            value,
+            default=self.default,
+            use_single_float=self.use_single_float,
+            use_bin_type=self.use_bin_type,
+            datetime=self.datetime,
+        )
+
+    def loadb(self, value: bytes) -> Any:
+        """Load value from bytes."""
+        return msgpack.unpackb(value, timestamp=3)

taskiq/serializers/msgpack_serializer.py

@@ -0,0 +1,34 @@
+from typing import Any, Callable, Optional
+
+from taskiq.abc.serializer import TaskiqSerializer
+
+try:
+    import orjson
+except ImportError:
+    orjson = None  # type: ignore
+
+
+class ORJSONSerializer(TaskiqSerializer):
+    """Taskiq serializer using orjson library."""
+
+    def __init__(
+        self,
+        default: Optional[Callable[[Any], Any]] = None,
+        option: Optional[int] = None,
+    ) -> None:
+        if orjson is None:
+            raise ImportError("orjson is not installed")
+        self.default = default
+        self.option = option
+
+    def dumpb(self, value: Any) -> bytes:
+        """Dump value to bytes."""
+        return orjson.dumps(
+            value,
+            default=self.default,
+            option=self.option,
+        )
+
+    def loadb(self, value: bytes) -> Any:
+        """Load value from bytes."""
+        return orjson.loads(value)

taskiq/serializers/orjson_serializer.py

@@ -45,7 +45,7 @@ def test_should_run_cron_str_offset() -> None:
 
 def test_should_run_cron_td_offset() -> None:
     offset = 2
-    hour = datetime.datetime.utcnow().hour + offset
+    hour = (datetime.datetime.utcnow().hour + offset) % 24
     delay = get_task_delay(
         ScheduledTask(
             task_name="",