fix(fastapi): resolve Optional body params to None for a Kafka tombstone

docs/docs/en/confluent/message.md

@@ -31,6 +31,13 @@ and its `raw_message` (the `confluent_kafka.Message`) exposes the rest through m
 * `#!python topic(): str`
 * `#!python value(): Optional[Union[str, bytes]]`
 
+!!! note
+    A record with a `None value()` is a Kafka tombstone, the delete marker on a compacted topic. `#!python msg.tombstone` is `#!python True` for it, keeping it distinct from an empty payload (`#!python b""`), and a `#!python None`-able body parameter of a **FastAPI** subscriber resolves to `#!python None` instead of failing validation.
+
+    A `#!python batch=True` subscriber does not populate the flag - per-record tombstones in a batch are left for a follow-up.
+
+    Publish one with `#!python await broker.publish(None, "topic", key=b"...")`.
+
 For example, if you would like to access the headers of an incoming message, you would do so like this:
 
 ```python hl_lines="1 6"

docs/docs/en/kafka/message.md

@@ -35,6 +35,13 @@ and its `raw_message` (the `aiokafka.ConsumerRecord`) contains the rest of the r
 * `#!python topic: str`
 * `#!python value: Optional[aiokafka.structs.VT]`
 
+!!! note
+    A record with a `None value` is a Kafka tombstone, the delete marker on a compacted topic. `#!python msg.tombstone` is `#!python True` for it, keeping it distinct from an empty payload (`#!python b""`), and a `#!python None`-able body parameter of a **FastAPI** subscriber resolves to `#!python None` instead of failing validation.
+
+    A `#!python batch=True` subscriber does not populate the flag - per-record tombstones in a batch are left for a follow-up.
+
+    Publish one with `#!python await broker.publish(None, "topic", key=b"...")`.
+
 For example, if you would like to access the headers of an incoming message, you would do so like this:
 
 ```python hl_lines="1 6"

faststream/_internal/fastapi/route.py

@@ -51,13 +51,13 @@ class StreamMessage(Request):
     scope: "dict[str, Any]"
     _cookies: "dict[str, Any]"
     _headers: "dict[str, Any]"  # type: ignore[assignment]
-    _body: Union["dict[str, Any]", list[Any]]  # type: ignore[assignment]
+    _body: Union["dict[str, Any]", list[Any], None]  # type: ignore[assignment]
     _query_params: "dict[str, Any]"  # type: ignore[assignment]
 
     def __init__(
         self,
         *,
-        body: Union["dict[str, Any]", list[Any]],
+        body: Union["dict[str, Any]", list[Any], None],
         headers: "dict[str, Any]",
         path: "dict[str, Any]",
     ) -> None:
@@ -172,9 +172,11 @@ async def parsed_consumer(message: "NativeMessage[Any]") -> Any:
         """Wrapper, that parser FastStream message to FastAPI compatible one."""
         body = await message.decode()
 
-        fastapi_body: dict[str, Any] | list[Any]
+        fastapi_body: dict[str, Any] | list[Any] | None
         if first_arg is not None:
-            if isinstance(body, dict):
+            if message.no_body:
+                fastapi_body, path = None, {}
+            elif isinstance(body, dict):
                 path = fastapi_body = body or {}
             elif isinstance(body, list):
                 fastapi_body, path = body, {}

faststream/confluent/message.py

@@ -59,6 +59,10 @@ def __init__(
         if not is_manual:
             self.committed = AckStatus.ACKED
 
+    @property
+    def tombstone(self) -> bool:
+        return self.no_body
+
     async def ack(self) -> None:
         """Acknowledge the Kafka message."""
         if self.is_manual and not self.committed:

faststream/confluent/parser.py

@@ -43,12 +43,14 @@ async def parse_message(
         """Parses a Kafka message."""
         headers = _parse_msg_headers(cast("_HeadersInput", message.headers() or ()))
 
-        body = message.value() or b""
+        value = message.value()
+        body = value or b""
         offset = message.offset()
         _, timestamp = message.timestamp()
 
         return KafkaMessage(
             body=body,
+            no_body=value is None,
             headers=headers,
             reply_to=headers.get("reply_to", ""),
             content_type=headers.get("content-type"),

faststream/kafka/message.py

@@ -61,6 +61,10 @@ def __init__(self, *args: Any, consumer: ConsumerProtocol, **kwargs: Any) -> Non
         self.consumer = consumer
         self.committed = AckStatus.ACKED
 
+    @property
+    def tombstone(self) -> bool:
+        return self.no_body
+
 
 class KafkaAckableMessage(KafkaMessage):
     def __init__(self, *args: Any, **kwargs: Any) -> None:

faststream/kafka/parser.py

@@ -46,9 +46,11 @@ async def parse_message(
     ) -> "StreamMessage[ConsumerRecord]":
         """Parses a Kafka message."""
         headers = {i: j.decode() for i, j in message.headers}
+        value = message.value
 
         return self.msg_class(
-            body=message.value or b"",
+            body=value or b"",
+            no_body=value is None,
             headers=headers,
             reply_to=headers.get("reply_to", ""),
             content_type=headers.get("content-type"),

faststream/message/message.py

@@ -43,9 +43,11 @@ def __init__(
         correlation_id: str | None = None,
         message_id: str | None = None,
         source_type: SourceType = SourceType.CONSUME,
+        no_body: bool = False,
     ) -> None:
         self.raw_message = raw_message
         self.body = body
+        self.no_body = no_body
         self.reply_to = reply_to
         self.content_type = content_type
         self.source_type = source_type

tests/brokers/base/fastapi.py

@@ -8,6 +8,7 @@
 from fastapi import BackgroundTasks, Depends, FastAPI, Header
 from fastapi.exceptions import RequestValidationError
 from fastapi.testclient import TestClient
+from pydantic import BaseModel
 
 from faststream import (
     Context as FSContext,
@@ -21,6 +22,10 @@
 from .basic import BaseTestcaseConfig
 
 
+class _Foo(BaseModel):
+    x: int
+
+
 @pytest.mark.asyncio()
 class FastAPITestcase(BaseTestcaseConfig[Any]):
     router_class: Any
@@ -270,6 +275,40 @@ async def subscriber(msg: StreamMessage) -> None:
         mock.assert_called_once_with(True)
 
 
+# NOTE: kafka/confluent only - other brokers have no tombstone concept.
+@pytest.mark.asyncio()
+class KafkaTombstoneFastAPILocalTestcase(BaseTestcaseConfig):
+    router_class: type[StreamRouter[BrokerUsecase]]
+
+    async def test_optional_body_resolves_to_none_for_tombstone(
+        self,
+        queue: str,
+    ) -> None:
+        router = self.router_class()
+        received: list[_Foo | None] = []
+
+        args, kwargs = self.get_subscriber_params(queue)
+
+        @router.subscriber(*args, **kwargs)
+        async def handler(msg: _Foo | None = None) -> None:
+            received.append(msg)
+
+        app = FastAPI()
+        app.include_router(router)
+
+        async with self.patch_broker(router.broker) as br:
+            with TestClient(app):
+                await br.publish(b'{"x": 5}', queue, key=b"k1")
+                await br.publish(None, queue, key=b"k2")
+
+                # an empty (non-null) body is not a tombstone and must
+                # still fail validation.
+                with pytest.raises(RequestValidationError):
+                    await br.publish(b"", queue, key=b"k3")
+
+        assert received == [_Foo(x=5), None]
+
+
 @pytest.mark.asyncio()
 class FastAPILocalTestcase(BaseTestcaseConfig[Any]):
     router_class: Any

tests/brokers/confluent/test_fastapi.py

@@ -4,9 +4,18 @@
 
 import pytest
 
+from faststream import Context
 from faststream.confluent import KafkaRouter
-from faststream.confluent.fastapi import KafkaRouter as StreamRouter
-from tests.brokers.base.fastapi import FastAPILocalTestcase, FastAPITestcase
+from faststream.confluent.fastapi import (
+    KafkaMessage,
+    KafkaRouter as StreamRouter,
+)
+from tests.brokers.base.fastapi import (
+    FastAPILocalTestcase,
+    FastAPITestcase,
+    KafkaTombstoneFastAPILocalTestcase,
+    _Foo,
+)
 
 from .basic import ConfluentMemoryTestcaseConfig, ConfluentTestcaseConfig
 
@@ -42,9 +51,45 @@ async def hello(msg: list[str]) -> Any:
         assert event.is_set()
         mock.assert_called_with(["hi"])
 
+    async def test_external_tombstone_resolves_to_none(
+        self,
+        queue: str,
+        event: asyncio.Event,
+    ) -> None:
+        router = self.router_class()
+        received: list[tuple[object, bytes | None]] = []
+
+        args, kwargs = self.get_subscriber_params(queue)
+
+        @router.subscriber(*args, **kwargs)
+        async def handler(
+            msg: _Foo | None = None,
+            raw: KafkaMessage = Context("message"),
+        ) -> None:
+            received.append((msg, raw.raw_message.value()))
+            if len(received) == 2:
+                event.set()
+
+        async with self.patch_broker(router.broker) as br:
+            await br.start()
+
+            await br.publish(b'{"x": 5}', queue, key=b"k1")
+            # bypass the encoder to prove this works for a tombstone
+            # produced by any client, not only faststream's publish(None)
+            await br._producer._producer.producer.send(topic=queue, key=b"k2", value=None)
+
+            await asyncio.wait_for(event.wait(), timeout=self.timeout)
+
+        assert (_Foo(x=5), b'{"x": 5}') in received
+        assert (None, None) in received
+
 
 @pytest.mark.confluent()
-class TestRouterLocal(ConfluentMemoryTestcaseConfig, FastAPILocalTestcase):
+class TestRouterLocal(
+    ConfluentMemoryTestcaseConfig,
+    FastAPILocalTestcase,
+    KafkaTombstoneFastAPILocalTestcase,
+):
     router_class = StreamRouter
     broker_router_class = KafkaRouter
 

tests/brokers/confluent/test_parser.py

@@ -1,5 +1,8 @@
+from unittest.mock import MagicMock
+
 import pytest
 
+from faststream.confluent.parser import AsyncConfluentParser
 from tests.brokers.base.parser import CustomParserTestcase
 
 from .basic import ConfluentTestcaseConfig
@@ -9,3 +12,29 @@
 @pytest.mark.confluent()
 class TestCustomParser(ConfluentTestcaseConfig, CustomParserTestcase):
     pass
+
+
+def _fake_message(value: bytes | None) -> MagicMock:
+    message = MagicMock()
+    message.value.return_value = value
+    message.headers.return_value = None
+    message.offset.return_value = 0
+    message.timestamp.return_value = (0, 0)
+    return message
+
+
+@pytest.mark.asyncio()
+@pytest.mark.confluent()
+@pytest.mark.parametrize(
+    ("value", "tombstone"),
+    ((None, True), (b"", False), (b"{}", False)),
+)
+async def test_parse_message_flags_only_a_null_value(
+    value: bytes | None,
+    tombstone: bool,
+) -> None:
+    parsed = await AsyncConfluentParser().parse_message(_fake_message(value))
+
+    assert parsed.no_body is tombstone
+    assert parsed.tombstone is tombstone
+    assert parsed.body == (value or b"")

tests/brokers/kafka/test_fastapi.py

@@ -4,9 +4,18 @@
 
 import pytest
 
+from faststream import Context
 from faststream.kafka import KafkaRouter
-from faststream.kafka.fastapi import KafkaRouter as StreamRouter
-from tests.brokers.base.fastapi import FastAPILocalTestcase, FastAPITestcase
+from faststream.kafka.fastapi import (
+    KafkaMessage,
+    KafkaRouter as StreamRouter,
+)
+from tests.brokers.base.fastapi import (
+    FastAPILocalTestcase,
+    FastAPITestcase,
+    KafkaTombstoneFastAPILocalTestcase,
+    _Foo,
+)
 
 from .basic import KafkaMemoryTestcaseConfig, KafkaTestcaseConfig
 
@@ -42,9 +51,47 @@ async def hello(msg: list[str]) -> Any:
         assert event.is_set()
         mock.assert_called_with(["hi"])
 
+    async def test_external_tombstone_resolves_to_none(
+        self,
+        queue: str,
+        event: asyncio.Event,
+    ) -> None:
+        router = self.router_class()
+        received: list[tuple[object, bytes | None]] = []
+
+        @router.subscriber(queue)
+        async def handler(
+            msg: _Foo | None = None,
+            raw: KafkaMessage = Context("message"),
+        ) -> None:
+            received.append((msg, raw.raw_message.value))
+            if len(received) == 2:
+                event.set()
+
+        async with router.broker as br:
+            await br.start()
+
+            await br.publish(b'{"x": 5}', queue, key=b"k1")
+            # bypass the encoder to prove this works for a tombstone
+            # produced by any client, not only faststream's publish(None)
+            await br._producer._producer.producer.send(
+                topic=queue,
+                key=b"k2",
+                value=None,
+            )
+
+            await asyncio.wait_for(event.wait(), timeout=3)
+
+        assert (_Foo(x=5), b'{"x": 5}') in received
+        assert (None, None) in received
+
 
 @pytest.mark.kafka()
-class TestRouterLocal(KafkaMemoryTestcaseConfig, FastAPILocalTestcase):
+class TestRouterLocal(
+    KafkaMemoryTestcaseConfig,
+    FastAPILocalTestcase,
+    KafkaTombstoneFastAPILocalTestcase,
+):
     router_class = StreamRouter
     broker_router_class = KafkaRouter