From 5555db755176722c485dc6258d8f226785b20a9a Mon Sep 17 00:00:00 2001 From: Arad Arang <50820293+aradng@users.noreply.github.com> Date: Mon, 13 Jul 2026 22:58:17 +0330 Subject: [PATCH] feat: shim the consumer-side tombstone fix too, until faststream releases it (#19) ag2ai/faststream#2933 (Optional[Model] = None resolving correctly for a real tombstone instead of crash-looping on required fields) is open, unmerged, no PyPI release. Producer-side already gets the equivalent treatment via _patch_real_tombstones - do the same here so both halves work today, not just once faststream ships something. Three patches, each guarded for idempotency: - AsyncConfluentParser.parse_message: tags the body with a dedicated TOMBSTONE sentinel (not a byte pattern - can't collide with real content) when the raw value is a genuine null. - AsyncConfluentParser.decode_message: maps TOMBSTONE to None. - faststream._internal.fastapi.route.build_faststream_to_fastapi_parser: a full replacement (it's a per-subscriber closure factory, not a patchable class method) that passes a real None through instead of wrapping it as {param_name: None} - that wrapping is what defeats FastAPI's own no-body shortcut and forces field-level validation on every tombstone. Bumps to 0.4.51. Co-authored-by: Claude Sonnet 5 --- fastloom/signals/kafka/depends.py | 142 +++++++++++++++++- .../fastloom-sdk/.claude-plugin/plugin.json | 2 +- pyproject.toml | 2 +- tests/kafka/test_tombstone.py | 34 +++++ 4 files changed, 177 insertions(+), 3 deletions(-) diff --git a/fastloom/signals/kafka/depends.py b/fastloom/signals/kafka/depends.py index 487e4ad..bdea645 100644 --- a/fastloom/signals/kafka/depends.py +++ b/fastloom/signals/kafka/depends.py @@ -1,26 +1,45 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from fastloom.meta import SelfSustaining from fastloom.signals.kafka.settings import KafkaSettings, KafkaSubscriptable if TYPE_CHECKING: from faststream.confluent.fastapi import KafkaRouter + from faststream.confluent.parser import AsyncConfluentParser from faststream.confluent.publisher.producer import ( AsyncConfluentFastProducerImpl, ) from faststream.confluent.response import KafkaPublishCommand +class _Tombstone: + """Sentinel marking a message body as a genuine null value - not a byte + pattern (`b"null"`, `b""`), so it can never collide with real content.""" + + __slots__ = () + + def __repr__(self) -> str: + return "TOMBSTONE" + + def __bool__(self) -> bool: + return False + + +TOMBSTONE = _Tombstone() + + def get_kafka_router(settings: KafkaSettings) -> KafkaRouter: # deferred: see docs/signals.md#ordering from faststream.confluent.fastapi import KafkaRouter + from faststream.confluent.parser import AsyncConfluentParser from faststream.confluent.publisher.producer import ( AsyncConfluentFastProducerImpl, ) _patch_real_tombstones(AsyncConfluentFastProducerImpl) + _patch_tombstone_consumption(AsyncConfluentParser) return KafkaRouter( settings.KAFKA_URI, @@ -67,6 +86,127 @@ async def publish( producer_cls._fastloom_real_tombstones = True +def _patch_tombstone_consumption( + parser_cls: type[AsyncConfluentParser], +) -> None: + # NOTE: consumer-side half of the same gap - a real tombstone decodes to + # b"" (parse_message()'s own `or b""`), so a typed Optional[Model] body + # param still crash-loops: FastAPI's body-solving flattens the model's + # own required fields rather than ever seeing "no body at all" + # (ag2ai/faststream#2933, open, no PyPI release either way). Tags the + # body with a dedicated sentinel at parse time, decodes it to None, and + # patches the fastapi bridge to pass real None through instead of + # wrapping it as {param_name: None} - the wrapping is what defeats + # FastAPI's own no-body shortcut in the first place. + if getattr(parser_cls, "_fastloom_tombstone_consumption", False): + return + + original_parse_message = parser_cls.parse_message + original_decode_message = parser_cls.decode_message + + async def parse_message(self: AsyncConfluentParser, message: Any): + parsed = await original_parse_message(self, message) + if message.value() is None: + parsed.body = TOMBSTONE + return parsed + + async def decode_message(self: AsyncConfluentParser, msg: Any): + if msg.body is TOMBSTONE: + return None + return await original_decode_message(self, msg) + + parser_cls.parse_message = parse_message + parser_cls.decode_message = decode_message + parser_cls._fastloom_tombstone_consumption = True + + _patch_fastapi_body_wrapping() + + +_fastapi_route_patched = False + + +def _patch_fastapi_body_wrapping() -> None: + global _fastapi_route_patched + if _fastapi_route_patched: + return + _fastapi_route_patched = True + + import inspect + from itertools import dropwhile + + import faststream._internal.fastapi.route as route + + def build_faststream_to_fastapi_parser( + *, + dependent: Any, + fastapi_config: Any, + context: Any, + response_field: Any, + response_model_include: Any, + response_model_exclude: Any, + response_model_by_alias: Any, + response_model_exclude_unset: Any, + response_model_exclude_defaults: Any, + response_model_exclude_none: Any, + ) -> Any: + assert dependent.call + + consume = route.make_fastapi_execution( + dependent=dependent, + fastapi_config=fastapi_config, + response_field=response_field, + response_model_include=response_model_include, + response_model_exclude=response_model_exclude, + response_model_by_alias=response_model_by_alias, + response_model_exclude_unset=response_model_exclude_unset, + response_model_exclude_defaults=response_model_exclude_defaults, + response_model_exclude_none=response_model_exclude_none, + ) + + dependencies_names = tuple(i.name for i in dependent.dependencies) + first_arg = next( + dropwhile( + lambda i: i in dependencies_names, + inspect.signature(dependent.call).parameters, + ), + None, + ) + + async def parsed_consumer(message: Any) -> Any: + body = await message.decode() + + fastapi_body: dict[str, Any] | list[Any] | None + if first_arg is not None: + if isinstance(body, dict): + path = fastapi_body = body or {} + elif isinstance(body, list): + fastapi_body, path = body, {} + elif body is None: + fastapi_body, path = None, {} + else: + path = fastapi_body = {first_arg: body} + + stream_message = route.StreamMessage( + body=fastapi_body, + headers={"context__": context, **message.headers}, + path={**path, **message.path}, + ) + else: + stream_message = route.StreamMessage( + body={}, + headers={"context__": context}, + path={}, + ) + + return await consume(stream_message, message) + + return parsed_consumer + + route.build_faststream_to_fastapi_parser = ( + build_faststream_to_fastapi_parser + ) + + class KafkaSubscriber(SelfSustaining): """Owns the shared FastStream KafkaRouter singleton.""" diff --git a/plugins/fastloom-sdk/.claude-plugin/plugin.json b/plugins/fastloom-sdk/.claude-plugin/plugin.json index c119040..b2c80e3 100644 --- a/plugins/fastloom-sdk/.claude-plugin/plugin.json +++ b/plugins/fastloom-sdk/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "fastloom-sdk", "description": "Skills for services that consume the fastloom library \u2014 scaffold new services, add routes and RabbitMQ subscribers, audit settings.", - "version": "0.4.50", + "version": "0.4.51", "author": { "name": "aradng" }, diff --git a/pyproject.toml b/pyproject.toml index 6f2aed7..0be78e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "fastloom" -version = "0.4.50" +version = "0.4.51" description = "Core package" authors = [] readme = "README.md" diff --git a/tests/kafka/test_tombstone.py b/tests/kafka/test_tombstone.py index 95c31ea..d69ab8b 100644 --- a/tests/kafka/test_tombstone.py +++ b/tests/kafka/test_tombstone.py @@ -3,6 +3,7 @@ from confluent_kafka import Message from faststream.confluent.fastapi import KafkaMessage +from pydantic import BaseModel async def test_publish_none_sends_a_real_null_value(kafka_subscriber): @@ -33,3 +34,36 @@ async def handler(msg: KafkaMessage) -> None: regular_value, tombstone_value = values assert regular_value == b"hello" assert tombstone_value is None + + +class _Foo(BaseModel): + x: int + + +async def test_optional_body_param_resolves_to_none_for_tombstone( + kafka_subscriber, +): + router = kafka_subscriber.router + received: list[_Foo | None] = [] + done = asyncio.Event() + + @router.subscriber( + "consumer-tombstone-test-topic", + group_id="consumer-tombstone-test", + auto_offset_reset="earliest", + ) + async def handler(msg: _Foo | None = None) -> None: + received.append(msg) + if len(received) == 2: + done.set() + + publisher = router.publisher("consumer-tombstone-test-topic") + await router.broker.start() + try: + await publisher.publish({"x": 5}, key=b"regular-message") + await publisher.publish(None, key=b"real-delete") + await asyncio.wait_for(done.wait(), timeout=15) + finally: + await router.broker.stop() + + assert received == [_Foo(x=5), None]