From d55590104e611453a90451fcc589c68a91d535cb Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 17 Jul 2026 11:33:57 +0300 Subject: [PATCH] refactor(message): name the envelope-managed header keys content-type and correlation_id were duplicated as bare string literals across three modules: envelope.py writes them, parser.py reads them back, and the subscriber's relay header-propagation strips them. Define them once in message.py (alongside the sibling DLQFailureReason wire contract) and repoint all five sites. - CONTENT_TYPE_HEADER / CORRELATION_ID_HEADER for the per-key read/write sites - ENVELOPE_MANAGED_HEADERS frozenset for the relay strip loop Behavior-preserving; adding a managed key is now a one-line edit instead of three. Full suite green (599 passed, coverage 100%), lint + ty clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- faststream_outbox/envelope.py | 12 +++++++----- faststream_outbox/message.py | 10 ++++++++++ faststream_outbox/parser/parser.py | 11 ++++++++--- faststream_outbox/subscriber/usecase.py | 4 ++-- 4 files changed, 27 insertions(+), 10 deletions(-) diff --git a/faststream_outbox/envelope.py b/faststream_outbox/envelope.py index dc4875b..0b532a4 100644 --- a/faststream_outbox/envelope.py +++ b/faststream_outbox/envelope.py @@ -5,6 +5,8 @@ from faststream.message import gen_cor_id from faststream.message.utils import encode_message +from faststream_outbox.message import CONTENT_TYPE_HEADER, CORRELATION_ID_HEADER + if TYPE_CHECKING: from fast_depends.library.serializer import SerializerProto @@ -28,25 +30,25 @@ def _encode_payload( """ payload, content_type = encode_message(body, serializer=serializer) out_headers: dict[str, str] = dict(headers or {}) - if "content-type" in out_headers and content_type and out_headers["content-type"] != content_type: + if CONTENT_TYPE_HEADER in out_headers and content_type and out_headers[CONTENT_TYPE_HEADER] != content_type: msg = ( - f"headers['content-type']={out_headers['content-type']!r} conflicts with the " + f"headers['content-type']={out_headers[CONTENT_TYPE_HEADER]!r} conflicts with the " f"encoder's output ({content_type!r}). Drop content-type from headers, or pass " "body as bytes if you need to label the payload yourself." ) raise ValueError(msg) if content_type: - out_headers["content-type"] = content_type + out_headers[CONTENT_TYPE_HEADER] = content_type # An explicit ``correlation_id`` kwarg used to lose silently to a # ``headers["correlation_id"]`` of a different value (the kwarg was dropped). # Treat a genuine mismatch as a conflict (like content-type above); otherwise the # kwarg wins when set, falling back to the header, then a fresh id (P2). - header_cid = out_headers.get("correlation_id") + header_cid = out_headers.get(CORRELATION_ID_HEADER) if correlation_id is not None and header_cid is not None and correlation_id != header_cid: msg = ( f"correlation_id={correlation_id!r} conflicts with headers['correlation_id']=" f"{header_cid!r}. Pass correlation_id one way, not both." ) raise ValueError(msg) - out_headers["correlation_id"] = correlation_id or header_cid or gen_cor_id() + out_headers[CORRELATION_ID_HEADER] = correlation_id or header_cid or gen_cor_id() return payload, out_headers diff --git a/faststream_outbox/message.py b/faststream_outbox/message.py index 223da85..412366c 100644 --- a/faststream_outbox/message.py +++ b/faststream_outbox/message.py @@ -36,6 +36,16 @@ DLQFailureReason = Literal["max_deliveries", "retry_terminal", "rejected"] +# Header keys the outbox envelope owns on a row's ``headers`` dict. ``_encode_payload`` +# (``envelope.py``) writes them, ``OutboxParser.parse_message`` (``parser.py``) reads +# them back, and the relay header-propagation (``_maybe_propagate_inbound_headers``) +# strips them before re-encode. Named here so those sites share one definition instead +# of repeating the string literals. +CONTENT_TYPE_HEADER = "content-type" +CORRELATION_ID_HEADER = "correlation_id" +ENVELOPE_MANAGED_HEADERS = frozenset({CONTENT_TYPE_HEADER, CORRELATION_ID_HEADER}) + + @dataclass(kw_only=True) class OutboxInnerMessage: """In-memory copy of a claimed outbox row, plus ack/nack/reject intent helpers. diff --git a/faststream_outbox/parser/parser.py b/faststream_outbox/parser/parser.py index 68358d5..5afde13 100644 --- a/faststream_outbox/parser/parser.py +++ b/faststream_outbox/parser/parser.py @@ -2,7 +2,12 @@ from faststream.message import decode_message -from faststream_outbox.message import OutboxInnerMessage, OutboxMessage +from faststream_outbox.message import ( + CONTENT_TYPE_HEADER, + CORRELATION_ID_HEADER, + OutboxInnerMessage, + OutboxMessage, +) class OutboxParser: @@ -12,9 +17,9 @@ async def parse_message(self, msg: OutboxInnerMessage) -> OutboxMessage: raw_message=msg, body=msg.payload, headers=headers, - content_type=headers.get("content-type"), + content_type=headers.get(CONTENT_TYPE_HEADER), message_id=str(msg.id), - correlation_id=headers.get("correlation_id", str(msg.id)), + correlation_id=headers.get(CORRELATION_ID_HEADER, str(msg.id)), # Set so ``SubscriberUsecase.__get_response_publisher`` (gated on a # truthy ``reply_to``) wires up the response publisher when a handler # returns ``OutboxResponse``. The inbound queue is the semantically diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index ee95b96..e6cabd9 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -39,7 +39,7 @@ from faststream.specification.schema import Message, Operation, SubscriberSpec from typing_extensions import override -from faststream_outbox.message import OutboxInnerMessage +from faststream_outbox.message import ENVELOPE_MANAGED_HEADERS, OutboxInnerMessage from faststream_outbox.parser.parser import OutboxParser from faststream_outbox.publisher.fake import OutboxFakePublisher from faststream_outbox.response import OutboxResponse @@ -1089,7 +1089,7 @@ def _maybe_propagate_inbound_headers( return propagated = dict(message.headers) if isinstance(result_msg, OutboxResponse): - for managed in ("content-type", "correlation_id"): + for managed in ENVELOPE_MANAGED_HEADERS: propagated.pop(managed, None) result_msg.headers = propagated