Merge branch 'develop' into dependabot/github_actions/pypa/gh-action-pypi-publish-1.9.0

Author: Lancetnik

.github/dependabot.yml

@@ -10,8 +10,16 @@ updates:
     directory: "/"
     schedule:
       interval: "weekly"
+    groups:
+      github-actions:
+        patterns:
+          - "*"
   # Python
   - package-ecosystem: "pip" # See documentation for possible values
     directory: "/" # Location of package manifests
     schedule:
       interval: "weekly"
+    groups:
+      pip:
+        patterns:
+          - "*"

.github/workflows/test.yml

@@ -40,7 +40,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12, "3.13"]
       fail-fast: false
 
     steps:

.gitignore

@@ -1,3 +1,5 @@
+uv.lock
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]
@@ -143,6 +145,8 @@ venv.bak/
 .dmypy.json
 dmypy.json
 
+.ruff_cache/
+
 # Pyre type checker
 .pyre/
 

pyproject.toml

@@ -13,19 +13,19 @@ authors = [
 
 keywords = ["taskiq", "tasks", "distributed", "async", "FastStream"]
 
-requires-python = ">=3.8"
+requires-python = ">=3.9"
 
 classifiers = [
     "Development Status :: 5 - Production/Stable",
     "License :: OSI Approved :: MIT License",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3 :: Only",
-    "Programming Language :: Python :: 3.8",
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "Operating System :: OS Independent",
     "Topic :: Software Development :: Libraries :: Python Modules",
     "Topic :: Software Development :: Libraries",
@@ -42,7 +42,7 @@ classifiers = [
 dynamic = ["version"]
 
 dependencies = [
-    "taskiq>=0.10.0,<1.0.0",
+    "taskiq>=0.11.0,<0.12.0",
     "faststream>=0.3.14,<0.6.0",
 ]
 
@@ -59,6 +59,10 @@ kafka = [
     "faststream[kafka]"
 ]
 
+confluent = [
+    "faststream[confluent]"
+]
+
 redis = [
     "faststream[redis]"
 ]
@@ -68,6 +72,7 @@ test = [
     "taskiq-faststream[nats]",
     "taskiq-faststream[rabbit]",
     "taskiq-faststream[kafka]",
+    "taskiq-faststream[confluent]",
     "taskiq-faststream[redis]",
 
     "coverage[toml]>=7.2.0,<8.0.0",
@@ -77,9 +82,9 @@ test = [
 dev = [
     "taskiq-faststream[test]",
 
-    "mypy>=1.8.0,<1.10.0",
-    "ruff==0.4.1",
-    "pre-commit >=3.6.0,<4.0.0",
+    "mypy>=1.8.0,<1.16.0",
+    "ruff==0.11.8",
+    "pre-commit >=3.6.0,<5.0.0",
 ]
 
 [project.urls]
@@ -102,7 +107,7 @@ exclude = [
 ]
 
 [tool.mypy]
-python_version = "3.8"
+python_version = "3.9"
 strict = true
 ignore_missing_imports = true
 allow_subclassing_any = true
@@ -120,12 +125,14 @@ known_third_party = ["faststream", "taskiq"]
 
 [tool.ruff]
 fix = true
-target-version = "py38"
+target-version = "py39"
 line-length = 88
-mccabe = { max-complexity = 10 }
 
+[tool.ruff.lint]
 # List of enabled rulsets.
 # See https://docs.astral.sh/ruff/rules/ for more information.
+mccabe = { max-complexity = 10 }
+
 select = [
     "E",   # Error
     "F",   # Pyflakes
@@ -162,8 +169,6 @@ ignore = [
     "D401",    # First line should be in imperative mood
     "D104",    # Missing docstring in public package
     "D100",    # Missing docstring in public module
-    "ANN102",  # Missing type annotation for self in method
-    "ANN101",  # Missing type annotation for argument
     "ANN401",  # typing.Any are disallowed in `**kwargs
     "PLR0913", # Too many arguments for function call
     "D106",    # Missing docstring in public nested class
@@ -172,7 +177,7 @@ ignore = [
 ]
 exclude = [".venv/"]
 
-[tool.ruff.per-file-ignores]
+[tool.ruff.lint.per-file-ignores]
 "tests/*" = [
     "S101", # Use of assert detected
     "S301", # Use of pickle detected
@@ -182,14 +187,14 @@ exclude = [".venv/"]
     "D101", # Missing docstring in public class
 ]
 
-[tool.ruff.pydocstyle]
+[tool.ruff.lint.pydocstyle]
 convention = "google"
 ignore-decorators = ["typing.overload"]
 
-[tool.ruff.pylint]
+[tool.ruff.lint.pylint]
 allow-magic-value-types = ["int", "str", "float"]
 
-[tool.ruff.flake8-bugbear]
+[tool.ruff.lint.flake8-bugbear]
 extend-immutable-calls = []
 
 [tool.pytest.ini_options]

scripts/lint.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 
 echo "Running ruff..."
-ruff taskiq_faststream tests --fix
+ruff check taskiq_faststream tests --fix
 
 echo "Running ruff formatter..."
 ruff format taskiq_faststream tests

taskiq_faststream/__about__.py

@@ -1,2 +1,3 @@
 """FastStream - taskiq integration to schedule FastStream tasks."""
-__version__ = "0.1.8"
+
+__version__ = "0.2.1"

taskiq_faststream/__init__.py

@@ -2,7 +2,7 @@
 from taskiq_faststream.scheduler import StreamScheduler
 
 __all__ = (
+    "AppWrapper",
     "BrokerWrapper",
     "StreamScheduler",
-    "AppWrapper",
 )

taskiq_faststream/broker.py

@@ -5,12 +5,12 @@
 import anyio
 from faststream.app import FastStream
 from faststream.types import SendableMessage
-from taskiq import AsyncBroker, BrokerMessage
+from taskiq import AsyncBroker
 from taskiq.acks import AckableMessage
 from taskiq.decor import AsyncTaskiqDecoratedTask
-from typing_extensions import TypeAlias, override
+from typing_extensions import TypeAlias
 
-from taskiq_faststream.serializer import PatchedSerializer
+from taskiq_faststream.formatter import PatchedFormatter, PathcedMessage
 from taskiq_faststream.types import ScheduledTask
 from taskiq_faststream.utils import resolve_msg
 
@@ -33,7 +33,7 @@ class BrokerWrapper(AsyncBroker):
 
     def __init__(self, broker: Any) -> None:
         super().__init__()
-        self.serializer = PatchedSerializer()
+        self.formatter = PatchedFormatter()
         self.broker = broker
 
     async def startup(self) -> None:
@@ -46,7 +46,7 @@ async def shutdown(self) -> None:
         await self.broker.close()
         await super().shutdown()
 
-    async def kick(self, message: BrokerMessage) -> None:
+    async def kick(self, message: PathcedMessage) -> None:  # type: ignore[override]
         """Call wrapped FastStream broker `publish` method."""
         await _broker_publish(self.broker, message)
 
@@ -66,8 +66,7 @@ async def listen(
             yield b""
             await anyio.sleep(60)
 
-    @override
-    def task(  # type: ignore[override]
+    def task(
         self,
         message: typing.Union[
             None,
@@ -76,7 +75,7 @@ def task(  # type: ignore[override]
             typing.Callable[[], typing.Awaitable[SendableMessage]],
         ] = None,
         *,
-        schedule: typing.List[ScheduledTask],
+        schedule: list[ScheduledTask],
         **kwargs: PublishParameters,
     ) -> "AsyncTaskiqDecoratedTask[[], None]":
         """Register FastStream scheduled task.
@@ -109,7 +108,7 @@ class AppWrapper(BrokerWrapper):
 
     def __init__(self, app: FastStream) -> None:
         super(BrokerWrapper, self).__init__()
-        self.serializer = PatchedSerializer()
+        self.formatter = PatchedFormatter()
         self.app = app
 
     async def startup(self) -> None:
@@ -122,7 +121,7 @@ async def shutdown(self) -> None:
         await self.app._shutdown()  # noqa: SLF001
         await super(BrokerWrapper, self).shutdown()
 
-    async def kick(self, message: BrokerMessage) -> None:
+    async def kick(self, message: PathcedMessage) -> None:  # type: ignore[override]
         """Call wrapped FastStream broker `publish` method."""
         assert (  # noqa: S101
             self.app.broker
@@ -132,11 +131,7 @@ async def kick(self, message: BrokerMessage) -> None:
 
 async def _broker_publish(
     broker: Any,
-    message: BrokerMessage,
+    message: PathcedMessage,
 ) -> None:
-    labels = message.labels
-    labels.pop("schedule", None)
-    async for msg in resolve_msg(
-        msg=labels.pop("message", message.message),
-    ):
-        await broker.publish(msg, **labels)
+    async for msg in resolve_msg(message.body):
+        await broker.publish(msg, **message.labels)

taskiq_faststream/formatter.py

@@ -0,0 +1,45 @@
+from dataclasses import dataclass
+from typing import Any
+
+from taskiq.abc.formatter import TaskiqFormatter
+from taskiq.message import TaskiqMessage
+
+
+@dataclass
+class PathcedMessage:
+    """DTO to transfer data to `broker.kick`."""
+
+    body: Any
+    labels: dict[str, Any]
+
+
+class PatchedFormatter(TaskiqFormatter):
+    """Default taskiq formatter."""
+
+    def dumps(  # type: ignore[override]
+        self,
+        message: TaskiqMessage,
+    ) -> PathcedMessage:
+        """
+        Dumps taskiq message to some broker message format.
+
+        :param message: message to send.
+        :return: Dumped message.
+        """
+        labels = message.labels
+        labels.pop("schedule", None)
+        labels.pop("schedule_id", None)
+
+        return PathcedMessage(
+            body=labels.pop("message", None),
+            labels=labels,
+        )
+
+    def loads(self, message: bytes) -> TaskiqMessage:
+        """
+        Loads json from message.
+
+        :param message: broker's message.
+        :return: parsed taskiq message.
+        """
+        raise NotImplementedError