4.0: Codec class, consolidated models, BaseConsumer + TransactionConsumer#60
Merged
Conversation
- Move all serialization/deserialization logic into rejected/codecs.py with stateless decode/encode/decode_avro/encode_avro functions - Rename message.py to models.py for the pydantic Message model - Split Consumer into BaseConsumer (framework plumbing) and Consumer (thin public API subclass preserving backward compatibility) - Add FunctionalConsumer that passes a Message model to prepare/process/finish instead of using self.body/self.content_type - Factor execute() into _pre_execute + _handle_execution + _run_consumer to share setup and error handling between Consumer and FunctionalConsumer Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
All pydantic models (Config, ConnectionConfig, ConnectionRef, ConsumerConfig, PrometheusConfig, StatsdConfig, StatsConfig) moved from config.py to models.py alongside the Message model. config.py retains the Settings wrapper class, the load() function, and re-exports all model classes for backward compatibility so existing `config_module.Config` references continue to work. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Set skip-magic-trailing-comma = true and isort.split-on-trailing-comma = false so ruff stops adding trailing commas everywhere. Reformatted all files with the new setting. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
- Result: IntEnum replacing module-level int constants (MESSAGE_ACK, MESSAGE_DROP, etc.) — backward-compat re-exports in data.py - Properties: pydantic model with from_pika() class method replacing the old __slots__-based Data subclass - InternalMessage: pydantic model with from_pika() class method replacing the old Message Data subclass - data.Message() and data.Properties() remain as backward-compatible factory functions that delegate to from_pika() - Measurement stays in data.py (mutable state, not a model) - Fixed fastavro references in consumer.py (codecs.fastavro) Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
- Annotate class attributes with ClassVar (HB_INTERVAL, STATE_CLOSED) - Fix on_open_error/on_closed: proper pika callback signatures instead of bare *args/**kwargs - Fix on_blocked/on_unblocked: use frame.Method[T] wrapper types - Fix on_consumer_cancelled/on_confirmation/on_qos_set: frame.Method[T] - Fix on_channel_closed: (Channel, Exception) per pika stubs, with isinstance check for ChannelClosed to extract reply_code/reply_text - Fix on_return: Channel type instead of int - Fix on_confirmation body: access method.method for the inner spec - Add on_channel_open return type annotation - Add explicit return None in _ssl_options - Add type: ignore for pika stub quirks on blocked/unblocked callbacks - Import pika.frame for Method wrapper type Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
pika doesn't ship py.typed or inline type stubs. The pika-stubs package provides type info for pyright/basedpyright but mypy needs an explicit override to suppress import-untyped errors. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
state.py: All methods, properties, __init__, and class attributes now fully annotated. Fixes mypy no-untyped-call errors in connection.py. connection.py: Restored type: ignore[arg-type] comments for three pika stub mismatches that mypy also flags (blocked/unblocked callback expects base Connection, ssl_options expects _DEFAULT sentinel). Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
codecs.py: - All functions fully annotated (params and return types) - Removed legacy plistlib compat (Python 3.11+ always has dumps/loads) - Fixed DictReader generic type arg (DictReader[str]) - Removed unused type: ignore comments on optional imports (mypy ignores those modules via overrides) pyproject.toml: - Added bs4, umsgpack, fastavro to mypy ignore_missing_imports - Added types-PyYAML and mypy to dev deps connection.py: - Restored type: ignore[arg-type] on 3 pika stub mismatches (needed because types-pika provides stubs that mypy DOES check) Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
log.py: - CorrelationFilter.__init__ and filter fully annotated - CorrelationAdapter with LoggerAdapter[Logger] generic param - __init__ and process fully annotated models.py: - Callbacks fields annotated as Callable[..., Any] - Message.connection and channel typed as Any (avoids circular import) - headers dict fully parameterized - Removed unused pika.channel TYPE_CHECKING import pyproject.toml: - Added yaml to mypy ignore_missing_imports (types-PyYAML not always available in all environments) Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Fully annotated for mypy strict mode: - state.py: all methods, properties, ClassVar attributes - connection.py: pika callback signatures with frame.Method[T], type: ignore[arg-type] for 3 pika stub quirks - codecs.py: all functions with param/return types, removed legacy plistlib compat - log.py: CorrelationFilter, CorrelationAdapter with LoggerAdapter generic param - models.py: Callbacks with Callable[..., Any], Message fields typed - exceptions.py: RejectedException.__init__/*args/**kwargs annotated, subclasses cleaned up (redundant __init__ removed) - utils.py: all functions annotated, str() wrapper on getattr return - measurement.py: Counter[str], dict type params, Generator return - config.py: Settings with full dict type params, ItemsView/KeysView - controller.py: run/signal handlers with FrameType, _mcp typed as MasterControlProgram | None - mcp.py: all 30+ methods annotated, ClassVar constants, signal handlers with FrameType, typed instance attributes - statsd.py: Client fully annotated including Callable for callback - mixins.py: GarbageCollectorMixin with ClassVar, hasattr guard for super().on_finish() - prometheus.py: dict type params, typing import added pyproject.toml: - mypy overrides for all untyped third-party modules - types-PyYAML and mypy added to dev deps Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Major restructuring for rejected 4.0. See PR description for details. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughReplaces legacy data/config/docs with Pydantic models and typed APIs, extracts async Codec and Connection modules, restructures Consumer/Process around ProcessingContext and TransactionConsumer, removes Sphinx docs/examples, and adds extensive typing, Avro/httpx support, and updated tests. Changes
Sequence Diagram(s)sequenceDiagram
participant Broker as Message Broker
participant Conn as Connection
participant Proc as Process
participant Codec as Codec
participant Consumer as Consumer
participant Ctx as ProcessingContext
Broker->>Conn: deliver(method, properties, body)
Conn->>Proc: on_delivery(method, properties, body)
Proc->>Proc: create ProcessingContext (ctx)
Proc->>Codec: decode(body, content_type, content_encoding, message_type)
Codec-->>Proc: decoded_payload
Proc->>Consumer: execute(ctx)
Consumer->>Consumer: prepare(ctx) / process(ctx) / finish(ctx)
alt success
Consumer-->>Proc: Result.MESSAGE_ACK
else error
Consumer-->>Proc: Result.* (CONSUMER_/MESSAGE_/PROCESSING_/UNHANDLED)
end
Proc->>Conn: ack/nack/publish based on Result
Conn->>Broker: basic_ack/basic_nack/basic_publish
sequenceDiagram
participant App as CLI/App
participant Ctrl as Controller
participant MCP as MasterControlProgram
participant Proc as Process
participant Conn as Connection
participant RMQ as RabbitMQ
App->>Ctrl: start(args, config)
Ctrl->>MCP: init(config)
MCP->>Proc: spawn processes for consumers
Proc->>Conn: Connection(config)
Conn->>RMQ: establish AsyncioConnection
RMQ-->>Conn: on_open -> channel created
Conn->>RMQ: basic_consume(queue)
RMQ-->>Proc: message delivery
Proc->>Proc: invoke_consumer(ctx)
Proc->>RMQ: ack/nack/publish
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rejected/utils.py (1)
55-56:⚠️ Potential issue | 🟡 MinorValidate percentile bounds before indexing.
kis currently unbounded;k == 0returns the last element andk > 100can raiseIndexError.Proposed fix
def percentile(values: list[float], k: int) -> float | None: @@ if not values: return None + if not 0 < k <= 100: + raise ValueError('k must be in the range 1..100') values.sort() index = (len(values) * (float(k) / 100)) - 1 return values[math.ceil(index)]🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/utils.py` around lines 55 - 56, Validate percentile input k before computing index: ensure k is within 1..100 (or explicitly define allowed range) and raise a clear ValueError if not; after computing index = (len(values) * (float(k) / 100)) - 1, clamp the resulting index into the valid range [0, len(values)-1] before calling math.ceil and indexing into values to avoid k==0 returning the last element and k>100 causing IndexError. Refer to the local variables index, values, and k in this fix.
🧹 Nitpick comments (2)
tests/test_data.py (1)
46-50: Add the exception-path check fortrack_duration.
Measurement.track_duration()appends in afinallyblock, but this suite
only exercises the happy path. A smallassertRaisescase here would lock in
the behavior that timings are still recorded when the wrapped block fails.Proposed test addition
def test_track_duration(self): with self.m.track_duration('timed'): pass self.assertEqual(len(self.m.durations['timed']), 1) self.assertGreaterEqual(self.m.durations['timed'][0], 0) + + def test_track_duration_on_exception(self): + with self.assertRaises(RuntimeError): + with self.m.track_duration('timed'): + raise RuntimeError('boom') + self.assertEqual(len(self.m.durations['timed']), 1) + self.assertGreaterEqual(self.m.durations['timed'][0], 0)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/test_data.py` around lines 46 - 50, Extend the existing test_track_duration in tests/test_data.py to also exercise the exception path: within an assertRaises (e.g., self.assertRaises(ValueError)) use self.m.track_duration('timed') as a context manager that raises inside the block, then after the exception assert that self.m.durations['timed'] length increased (or contains an entry) and that the recorded duration is >= 0, thereby verifying Measurement.track_duration's finally block behavior; locate the test by the test_track_duration method and the code under test by Measurement.track_duration and self.m.durations.pyproject.toml (1)
133-137: Clarify the Ruff glob pattern for repository-wide suppression.The pattern
"*.py" = ["S101", "S301"]is ambiguous in Ruff. Without a pinned version, it may behave unexpectedly—prior to v0.5.0,*.pymatched recursively; v0.5.0+ uses gitignore semantics where*doesn't cross directories. Since your codebase uses explicit**/*.pyfor other directories (tests/**/*.py,utils/**/*.py), use**/*.pyhere as well for clarity and consistent behavior across Ruff versions.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pyproject.toml` around lines 133 - 137, The Ruff suppression pattern `"*.py" = ["S101", "S301"]` is ambiguous across Ruff versions; update that key to use a recursive glob `"**/*.py"` so it matches files across all directories consistently (matching how `tests/**/*.py` and `utils/**/*.py` are used) and replace the `"*.py"` entry with `"**/*.py"` in pyproject.toml to ensure repository-wide suppression behaves the same regardless of Ruff version.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@README.md`:
- Around line 44-53: The README example uses the wrong Consumer.process
signature — change the method on class Test (class Test(consumer.Consumer)) from
async def process(self, message: models.Message) to the no-arg async def
process(self) and update the body to reference the incoming message via the
consumer instance (e.g. use self.message.body) so the example matches the
consumer.Consumer API used in tests; if you intended an explicit message
parameter instead, switch to TransactionConsumer in the example.
In `@rejected/codecs.py`:
- Around line 113-114: The block that calls pickle.loads when content_type is in
PICKLE_MIME_TYPES is unsafe for untrusted data; change it to only use
pickle.loads behind an explicit, configurable opt-in (e.g., an
ENABLE_PICKLE_DESERIALIZATION boolean or runtime setting) and fail with a clear
error/log if deserialization is requested but the flag is false, document the
trust assumption for PICKLE_MIME_TYPES, and wrap the pickle.loads call in a
try/except to log and rethrow deserialization errors; alternatively implement or
call a safe deserializer (JSON/msgpack) for untrusted sources and reserve
pickle.loads for trusted/internal flows only.
In `@rejected/connection.py`:
- Around line 327-378: The docstring in the _ssl_options property contains a
typo ("prototcol"); update the expected ssl_options list to spell "protocol"
correctly in the docstring for the _ssl_options method (referencing the
_ssl_options property and config.ssl_options) so the documentation is accurate.
In `@rejected/exceptions.py`:
- Around line 26-45: The exception class's __init__/__str__/__repr__ currently
stores non-string args in self.value and then calls .format() in __str__, which
can raise a second exception; update __init__ to ensure self.value is always a
string (coerce args[0] to str when used as the default value) and/or change
__str__ to only call self.value.format(...) when isinstance(self.value, str)
(otherwise return str(self.value)); reference the __init__, __str__, __repr__,
self.value and self.args symbols when making this safe-change so formatting
never attempts to call .format on a non-string.
---
Outside diff comments:
In `@rejected/utils.py`:
- Around line 55-56: Validate percentile input k before computing index: ensure
k is within 1..100 (or explicitly define allowed range) and raise a clear
ValueError if not; after computing index = (len(values) * (float(k) / 100)) - 1,
clamp the resulting index into the valid range [0, len(values)-1] before calling
math.ceil and indexing into values to avoid k==0 returning the last element and
k>100 causing IndexError. Refer to the local variables index, values, and k in
this fix.
---
Nitpick comments:
In `@pyproject.toml`:
- Around line 133-137: The Ruff suppression pattern `"*.py" = ["S101", "S301"]`
is ambiguous across Ruff versions; update that key to use a recursive glob
`"**/*.py"` so it matches files across all directories consistently (matching
how `tests/**/*.py` and `utils/**/*.py` are used) and replace the `"*.py"` entry
with `"**/*.py"` in pyproject.toml to ensure repository-wide suppression behaves
the same regardless of Ruff version.
In `@tests/test_data.py`:
- Around line 46-50: Extend the existing test_track_duration in
tests/test_data.py to also exercise the exception path: within an assertRaises
(e.g., self.assertRaises(ValueError)) use self.m.track_duration('timed') as a
context manager that raises inside the block, then after the exception assert
that self.m.durations['timed'] length increased (or contains an entry) and that
the recorded duration is >= 0, thereby verifying Measurement.track_duration's
finally block behavior; locate the test by the test_track_duration method and
the code under test by Measurement.track_duration and self.m.durations.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 0780a2f3-6424-464c-8fd7-e4d9fbf5b16b
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (37)
LICENSEREADME.mddocs/Makefiledocs/conf.pydocs/configuration.rstdocs/consumer_howto.rstdocs/history.rstdocs/index.rstexamples.pypyproject.tomlrejected/__init__.pyrejected/codecs.pyrejected/config.pyrejected/connection.pyrejected/consumer.pyrejected/controller.pyrejected/data.pyrejected/exceptions.pyrejected/log.pyrejected/mcp.pyrejected/measurement.pyrejected/mixins.pyrejected/models.pyrejected/process.pyrejected/prometheus.pyrejected/state.pyrejected/statsd.pyrejected/testing.pyrejected/utils.pytests/test_consumer.pytests/test_controller.pytests/test_data.pytests/test_mcp.pytests/test_process.pytests/test_statsd.pytests/test_testing.pyutils/test_generator.py
💤 Files with no reviewable changes (9)
- docs/index.rst
- docs/conf.py
- docs/history.rst
- docs/configuration.rst
- examples.py
- docs/consumer_howto.rst
- docs/Makefile
- utils/test_generator.py
- rejected/data.py
- Regenerate uv.lock without pypi.aweber.io URLs that caused DNS failures in GitHub Actions - Fix README example to use correct Consumer.process() signature (no-arg, access message via self.body) - Fix typo in connection.py docstring: prototcol -> protocol - Coerce exception value to str in RejectedException.__init__ to prevent .format() from failing on non-string args Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Owner
Author
PR Monitor SummaryCI/CD Fixes
CodeRabbit Review Feedback
All 4 review threads have been resolved. Waiting for CI re-run and CodeRabbit re-review. |
Add `from __future__ import annotations` to connection.py so that pika.frame.Method[pika.spec.Connection.Blocked] and similar type annotations are treated as strings rather than evaluated at class definition time. Without this, Python 3.11-3.13 raise TypeError because pika.frame.Method does not support __class_getitem__. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add `from __future__ import annotations` to codecs.py so that csv.DictReader[str] return type annotation is treated as a string rather than evaluated at runtime. csv.DictReader became subscriptable in Python 3.12 but this project supports 3.11+. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Owner
Author
PR Monitor - Final SummaryAll CI checks are now passing across Python 3.11, 3.12, 3.13, and 3.14. All CodeRabbit review threads have been addressed and resolved. Commits pushed
CodeRabbit feedback
|
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rejected/connection.py`:
- Around line 65-82: The shutdown flow currently only flips state when no
channel exists, allowing on_open/on_channel_open/on_ready to continue startup
and call into processing; update shutdown() (and mirror the change in the other
startup/teardown paths mentioned) so that if self.is_shutting_down is set and
self.channel is None you close the underlying connection instead of only setting
STATE_SHUTTING_DOWN, and additionally add early-return guards in on_open,
on_channel_open and on_ready to short-circuit (no-op) when self.is_shutting_down
is True; reference the methods shutdown(), on_open(), on_channel_open(),
on_ready(), the flag is_shutting_down, the attribute channel, and
set_state/STATE_SHUTTING_DOWN when applying the change.
- Around line 164-178: The on_channel_closed handler currently deletes the
self.channel attribute which causes later checks like "if self.channel:" to
raise AttributeError; instead, update on_channel_closed (the method
on_channel_closed) to set self.channel = None so the attribute remains present
and correctly typed (Channel | None) for subsequent teardown logic and guards to
work.
- Around line 343-378: The method building SSL options should also respect a
top-level self.config.ssl boolean and must pass server_hostname into
pika.SSLOptions; update the logic in the SSL builder (references:
self.config.ssl_options, self.config.ssl, and the returned pika.SSLOptions) so
that if ssl_options is empty but self.config.ssl is truthy you still create an
SSLContext; when protocol == ssl.PROTOCOL_TLS_CLIENT use
ssl.create_default_context() to get secure defaults, otherwise use
ssl.SSLContext(protocol); keep existing
load_verify_locations/load_cert_chain/set_ciphers work, and finally return
pika.SSLOptions(context=context, server_hostname=self.config.host or
self.config.hostname) so hostname verification occurs during the TLS handshake.
- Around line 260-268: The handler on_consumer_cancelled currently uses an elif
that prevents closing self.channel when the broker cancels a consumer and
self.is_shutting_down is False; change the elif to an independent if so that
after calling self.set_state(self.STATE_SHUTTING_DOWN) the code still checks and
closes the channel (call self.channel.close() when self.channel is truthy),
ensuring the connection doesn’t remain half-shutdown.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5e5ed4e8-4152-4a33-b600-df4c04dd7f06
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (3)
README.mdrejected/connection.pyrejected/exceptions.py
🚧 Files skipped from review as they are similar to previous changes (2)
- rejected/exceptions.py
- README.md
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (4)
rejected/connection.py (4)
67-83:⚠️ Potential issue | 🟠 MajorShutdown during CONNECTING state doesn't prevent startup completion.
When
shutdown()is called before a channel exists (e.g., during CONNECTING), the method only sets the state but doesn't close the underlying connection. Theon_open()andon_channel_open()callbacks lack guards foris_shutting_down, allowing startup to complete andon_ready()to fire after a stop was requested.Add early-return guards in
on_openandon_channel_openwhenis_shutting_downis true, or close the connection here when no channel exists.Suggested fix
def shutdown(self) -> None: if self.is_shutting_down: LOGGER.debug('Connection %s is already shutting down', self.name) return self.set_state(self.STATE_SHUTTING_DOWN) LOGGER.debug('Connection %s is shutting down', self.name) if self.is_active and self.channel: LOGGER.debug( 'Connection %s is sending a Basic.Cancel to RabbitMQ', self.name, ) self.channel.basic_cancel( self.consumer_tag, self.on_consumer_cancelled ) elif self.channel: self.channel.close() + elif self.connection: + self.connection.close()And add guards in callbacks:
def on_open(self, connection: asyncio_connection.AsyncioConnection) -> None: if self.is_shutting_down: connection.close() return # ... rest of method def on_channel_open(self, channel: pika.channel.Channel) -> None: if self.is_shutting_down: channel.close() return # ... rest of method🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/connection.py` around lines 67 - 83, The shutdown flow can race with startup: shutdown() sets is_shutting_down but if called before a channel exists the connection is never closed and on_open/on_channel_open proceed to complete startup and call on_ready; fix by adding early-return guards that close resources when shutdown is in progress and/or closing the connection when no channel exists in shutdown(): in shutdown() detect the no-channel case and close the underlying connection, and in on_open(self, connection) and on_channel_open(self, channel) immediately close the passed connection/channel and return if self.is_shutting_down to prevent completing startup and firing on_ready.
345-380:⚠️ Potential issue | 🟠 MajorPass
server_hostnametopika.SSLOptionsfor proper TLS hostname verification.
pika.SSLOptionsrequiresserver_hostnamefor hostname verification during the TLS handshake. Without it, the connection is vulnerable to MITM attacks. Additionally, when usingssl.PROTOCOL_TLS_CLIENT, consider usingssl.create_default_context()which provides secure defaults including certificate validation and hostname checking.Suggested fix
- if not self.config.ssl_options: + if not self.config.ssl_options and not getattr( + self.config, 'ssl', False + ): return None protocol = self.config.ssl_options.get( 'protocol', ssl.PROTOCOL_TLS_CLIENT ) if isinstance(protocol, str): protocol = getattr(ssl, protocol) - context = ssl.SSLContext(protocol) + if protocol == ssl.PROTOCOL_TLS_CLIENT: + context = ssl.create_default_context() + else: + context = ssl.SSLContext(protocol) # ... CA and cert loading unchanged ... - return pika.SSLOptions(context=context) + return pika.SSLOptions( + context=context, + server_hostname=self.config.ssl_options.get( + 'server_hostname', self.config.host + ), + )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/connection.py` around lines 345 - 380, The SSL options builder is missing server_hostname when constructing pika.SSLOptions and should use a secure default SSLContext; update the code that builds the context (where protocol/SSLContext is created and returned via pika.SSLOptions) to create the context with ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH) or, if using SSLContext(protocol), enable hostname checking and set verify_mode, then pass the peer hostname from self.config.ssl_options (e.g. server_hostname or host) into pika.SSLOptions(..., server_hostname=...) so TLS certificate hostname verification is performed.
262-270:⚠️ Potential issue | 🟠 MajorChange
eliftoifto ensure channel closes after broker-initiated cancels.When the broker initiates a cancel,
is_shutting_downisFalse, so line 268 sets the state but theelifon line 269 is skipped, leaving the channel open. These should be independent checks.Suggested fix
def on_consumer_cancelled( self, _method: pika.frame.Method[pika.spec.Basic.CancelOk] ) -> None: """Invoked by pika when a ``Basic.CancelOk`` is received.""" LOGGER.info('Connection %s consumer has been cancelled', self.name) if not self.is_shutting_down: self.set_state(self.STATE_SHUTTING_DOWN) - elif self.channel: + if self.channel: self.channel.close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/connection.py` around lines 262 - 270, The handler on_consumer_cancelled currently uses an elif so when is_shutting_down is False we set_state(self.STATE_SHUTTING_DOWN) but skip closing the channel; change the elif self.channel: to an independent if self.channel: so the channel.close() runs regardless of shutdown state. Update the block referencing on_consumer_cancelled, is_shutting_down, set_state, STATE_SHUTTING_DOWN and channel.close accordingly.
179-179:⚠️ Potential issue | 🔴 CriticalDon't delete
self.channel; set it toNoneinstead.
self.channelis typed asChannel | Noneand checked withif self.channel:elsewhere (lines 74, 82, 269). Deleting the attribute will causeAttributeErroron subsequent access. Set it toNoneto maintain type consistency.Suggested fix
- del self.channel + self.channel = None🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/connection.py` at line 179, Replace the deletion of the channel attribute with a None assignment: wherever the code currently does "del self.channel" (e.g., in the connection teardown/cleanup method), change it to "self.channel = None" so the attribute remains present and matches its declared type (Channel | None) for subsequent "if self.channel:" checks in the class.
🧹 Nitpick comments (1)
rejected/connection.py (1)
223-223: Consider settingself.connection = Noneinstead of deleting.Similar to the
self.channelissue, deletingself.connectioncan causeAttributeErrorif any code path accesses it after failure. While the connection failure should trigger process shutdown, setting toNoneis safer and more consistent with the typed attribute pattern.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/connection.py` at line 223, Replace the deletion of the connection attribute with setting it to None: change the statement that does "del self.connection" to "self.connection = None" so subsequent accesses won't raise AttributeError; also ensure the owning class's attribute (self.connection) is typed/treated as Optional (and mirror the same pattern used for self.channel) so callers can safely check for None.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rejected/codecs.py`:
- Around line 103-109: When content_type equals AVRO_DATUM_MIME_TYPE but
required prerequisites (fastavro module or message_type) are missing, fail fast
instead of falling through; in the decode/encode paths where the current
conditional checks fastavro is not None and message_type (around the logic that
calls self._avro_schema and fastavro.schemaless_reader), raise a clear exception
(e.g., ValueError or a custom CodecError) indicating that Avro support is
unavailable and which prerequisite is missing (fastavro or message_type). Do the
same for the corresponding encode branch (the similar block referenced at lines
142-147) so both encode and decode consistently error when Avro prerequisites
are not satisfied.
- Around line 100-101: Replace uses of zlib for gzip handling: change the
decompression in the content_encoding == 'gzip' branch (where body =
zlib.decompress(body)) to use gzip.decompress(body) and likewise replace the
compression call (zlib.compress(body)) with gzip.compress(body); add import gzip
to the module imports so RFC 1952-compliant gzip streams are used; locate the
changes in the codecs.py functions that handle content_encoding and
response/body compression.
- Around line 180-184: The async HTTP client created lazily in
_load_http_schema() is never closed because the shutdown flow doesn't await
Codec.close(); update the shutdown/cleanup coroutine in the process module to
await codec.close() for every instantiated codec (or provide an async close_all
that iterates and awaits codec.close()), ensuring the Codec.close() async method
is called for codecs that created an httpx.AsyncClient so the HTTP client is
properly aclosed and set to None.
- Around line 98-124: The decode logic in codecs.py can raise raw exceptions
from bz2.decompress, zlib.decompress, pickle.loads, plistlib.loads,
yaml.safe_load, etc., which bypass the caller's expected DecodeError handling;
wrap each decoding branch (content_encoding handling and the branches that call
bz2.decompress, zlib.decompress, pickle.loads for PICKLE_MIME_TYPES,
plistlib.loads for 'application/x-plist', yaml.safe_load for YAML_MIME_TYPES,
and any other loaders like _load_json/_load_msgpack/_load_csv/_load_bs4) in a
try/except that catches the implementation-specific exceptions and re-raises a
unified DecodeError (preserving original exception info/message) so callers
always receive DecodeError; keep the existing fastavro path using
self._avro_schema and fastavro.schemaless_reader but also convert its errors to
DecodeError if needed.
---
Duplicate comments:
In `@rejected/connection.py`:
- Around line 67-83: The shutdown flow can race with startup: shutdown() sets
is_shutting_down but if called before a channel exists the connection is never
closed and on_open/on_channel_open proceed to complete startup and call
on_ready; fix by adding early-return guards that close resources when shutdown
is in progress and/or closing the connection when no channel exists in
shutdown(): in shutdown() detect the no-channel case and close the underlying
connection, and in on_open(self, connection) and on_channel_open(self, channel)
immediately close the passed connection/channel and return if
self.is_shutting_down to prevent completing startup and firing on_ready.
- Around line 345-380: The SSL options builder is missing server_hostname when
constructing pika.SSLOptions and should use a secure default SSLContext; update
the code that builds the context (where protocol/SSLContext is created and
returned via pika.SSLOptions) to create the context with
ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH) or, if using
SSLContext(protocol), enable hostname checking and set verify_mode, then pass
the peer hostname from self.config.ssl_options (e.g. server_hostname or host)
into pika.SSLOptions(..., server_hostname=...) so TLS certificate hostname
verification is performed.
- Around line 262-270: The handler on_consumer_cancelled currently uses an elif
so when is_shutting_down is False we set_state(self.STATE_SHUTTING_DOWN) but
skip closing the channel; change the elif self.channel: to an independent if
self.channel: so the channel.close() runs regardless of shutdown state. Update
the block referencing on_consumer_cancelled, is_shutting_down, set_state,
STATE_SHUTTING_DOWN and channel.close accordingly.
- Line 179: Replace the deletion of the channel attribute with a None
assignment: wherever the code currently does "del self.channel" (e.g., in the
connection teardown/cleanup method), change it to "self.channel = None" so the
attribute remains present and matches its declared type (Channel | None) for
subsequent "if self.channel:" checks in the class.
---
Nitpick comments:
In `@rejected/connection.py`:
- Line 223: Replace the deletion of the connection attribute with setting it to
None: change the statement that does "del self.connection" to "self.connection =
None" so subsequent accesses won't raise AttributeError; also ensure the owning
class's attribute (self.connection) is typed/treated as Optional (and mirror the
same pattern used for self.channel) so callers can safely check for None.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 100e85b3-451a-4e21-a55f-482fad203f19
📒 Files selected for processing (2)
rejected/codecs.pyrejected/connection.py
connection.py: - Add early-return guards in on_open/on_channel_open when is_shutting_down, preventing startup from continuing after shutdown is requested during CONNECTING state - Close underlying connection when shutdown() is called before a channel exists (no channel to cancel/close) - Replace `del self.channel` with `self.channel = None` in on_channel_closed to avoid AttributeError in teardown guards - Change elif to if in on_consumer_cancelled so channel always closes after broker-initiated consumer cancels - Use ssl.create_default_context() for PROTOCOL_TLS_CLIENT to get secure defaults (cert validation, hostname checking) - Check self.config.ssl bool to enable TLS without explicit ssl_options dict - Pass server_hostname to pika.SSLOptions for proper TLS hostname verification codecs.py: - Replace zlib with gzip module for gzip content_encoding to comply with RFC 1952 (proper gzip framing, not raw zlib) - Wrap entire decode body in try/except to normalize all failures to DecodeError consistently - Fail fast with clear error when Avro content_type is used but fastavro or message_type is missing (both encode/decode) process.py: - Wire codec.close() into stop_consumer so httpx.AsyncClient is properly closed during shutdown - execute() now returns None, result stored on ctx.result consumer.py: - execute() return type changed from Result to None; result is now set on ctx.result testing.py, test_consumer.py: - Update callers to read ctx.result instead of execute() return Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Owner
Author
|
🤖 This comment was posted by Claude on behalf of @gmr PR Monitor SummaryResponse to CodeRabbit's Review (8 threads addressed)connection.py (4 threads):
codecs.py (4 threads):
Additional Changes (included in this push)
|
There was a problem hiding this comment.
🧹 Nitpick comments (3)
rejected/codecs.py (1)
263-272: Consider adding a timeout or retry for HTTP schema fetches.The HTTP schema loading uses a 30-second timeout but has no retry logic. For transient network failures, the first message requiring a schema will fail with
DecodeError. This may be acceptable for the use case, but consider whether a retry with backoff would improve resilience.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/codecs.py` around lines 263 - 272, The HTTP schema fetch in the method using self._http_client (creating httpx.AsyncClient(timeout=30) and awaiting response = await self._http_client.get(uri)) needs retry/backoff to handle transient failures: wrap the GET in a small retry loop (e.g. 2–3 attempts) with exponential backoff delays, catch transient exceptions (httpx.RequestError, timeouts) and non-200 responses, and only raise DecodeError (for the message_type) after all retries fail; ensure the client creation (self._http_client) remains reused and the retry logic targets the GET call and response.status_code handling.rejected/testing.py (1)
228-250: Consider handlingMESSAGE_REQUEUEresult in test assertions.The
process_messagemethod raises exceptions forCONSUMER_EXCEPTION,MESSAGE_EXCEPTION,PROCESSING_EXCEPTION, andUNHANDLED_EXCEPTION, butMESSAGE_REQUEUEandMESSAGE_DROPsilently return the measurement. If a test expects an error but gets a requeue, the test would pass without indication. This may be intentional for the testing use case, but consider documenting this behavior.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/testing.py` around lines 228 - 250, The test currently ignores models.Result.MESSAGE_REQUEUE (and MESSAGE_DROP) after await self.consumer.execute(ctx) which can hide failures; update the post-execute result handling (the block checking ctx.result) to explicitly handle models.Result.MESSAGE_REQUEUE (and optionally models.Result.MESSAGE_DROP) by raising a clear test exception (e.g., ConsumerRequeueException or AssertionError with context) or by asserting the expected requeue/drop behavior so a requeue doesn't silently make the test pass — locate the branch around ctx.result checks in the same function where _capture_log and self.consumer._log_exception are set and add the new case(s) there.rejected/connection.py (1)
20-26: State constantSTATE_CLOSED = 0x08intentionally aliasesSTATE_STOPPED.The
STATE_CLOSEDconstant shares the same value (0x08) as the parent'sSTATE_STOPPED. This appears intentional since theSTATESdict overrides the 0x08 label with'Closed'for connection-specific semantics, andis_closedcorrectly delegates tois_stopped. However, this coupling could cause confusion if the parent class changes.Consider using a distinct value (e.g.,
0x09) or documenting this aliasing explicitly:# Intentionally aliased to STATE_STOPPED for is_stopped/is_closed equivalence STATE_CLOSED: typing.ClassVar[int] = state.State.STATE_STOPPED🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/connection.py` around lines 20 - 26, The Connection class defines STATE_CLOSED as 0x08 which aliases the parent state's STATE_STOPPED; update the declaration of STATE_CLOSED in Connection to avoid accidental divergence or clarify intent: either assign it explicitly from the parent (STATE_CLOSED = state.State.STATE_STOPPED) with an inline comment like "Intentionally aliased to STATE_STOPPED for is_stopped/is_closed equivalence", or choose a distinct value (e.g., 0x09) and update STATES accordingly; ensure references to Connection.STATES, is_closed and is_stopped remain consistent after the change.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@rejected/codecs.py`:
- Around line 263-272: The HTTP schema fetch in the method using
self._http_client (creating httpx.AsyncClient(timeout=30) and awaiting response
= await self._http_client.get(uri)) needs retry/backoff to handle transient
failures: wrap the GET in a small retry loop (e.g. 2–3 attempts) with
exponential backoff delays, catch transient exceptions (httpx.RequestError,
timeouts) and non-200 responses, and only raise DecodeError (for the
message_type) after all retries fail; ensure the client creation
(self._http_client) remains reused and the retry logic targets the GET call and
response.status_code handling.
In `@rejected/connection.py`:
- Around line 20-26: The Connection class defines STATE_CLOSED as 0x08 which
aliases the parent state's STATE_STOPPED; update the declaration of STATE_CLOSED
in Connection to avoid accidental divergence or clarify intent: either assign it
explicitly from the parent (STATE_CLOSED = state.State.STATE_STOPPED) with an
inline comment like "Intentionally aliased to STATE_STOPPED for
is_stopped/is_closed equivalence", or choose a distinct value (e.g., 0x09) and
update STATES accordingly; ensure references to Connection.STATES, is_closed and
is_stopped remain consistent after the change.
In `@rejected/testing.py`:
- Around line 228-250: The test currently ignores models.Result.MESSAGE_REQUEUE
(and MESSAGE_DROP) after await self.consumer.execute(ctx) which can hide
failures; update the post-execute result handling (the block checking
ctx.result) to explicitly handle models.Result.MESSAGE_REQUEUE (and optionally
models.Result.MESSAGE_DROP) by raising a clear test exception (e.g.,
ConsumerRequeueException or AssertionError with context) or by asserting the
expected requeue/drop behavior so a requeue doesn't silently make the test pass
— locate the branch around ctx.result checks in the same function where
_capture_log and self.consumer._log_exception are set and add the new case(s)
there.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 414831a8-bf2e-45ce-9c25-aa5e03c54251
📒 Files selected for processing (6)
rejected/codecs.pyrejected/connection.pyrejected/consumer.pyrejected/process.pyrejected/testing.pytests/test_consumer.py
🚧 Files skipped from review as they are similar to previous changes (1)
- tests/test_consumer.py
codecs.py: HTTP schema fetch retries 3 times with exponential backoff, catching httpx.RequestError and non-200 responses before raising DecodeError. connection.py: STATE_CLOSED explicitly assigned from state.State.STATE_STOPPED with comment clarifying the intentional alias for is_closed/is_stopped equivalence. testing.py: process_message now raises AssertionError on MESSAGE_REQUEUE (unexpected in tests). MESSAGE_DROP returns normally (valid test outcome). Uses match/case for clarity. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (1)
rejected/connection.py (1)
233-240:⚠️ Potential issue | 🟠 MajorDon't delete
self.connectionfrom object state.This recreates the same teardown hazard as deleting
self.channel: later paths still assume the attribute exists. Afteron_failure(), a latershutdown()will reach Line 87 and raiseAttributeErrorinstead of becoming a no-op or clean close.Minimal fix
def on_failure(self) -> None: LOGGER.info('Connection failure, terminating connection') self.set_state(self.STATE_CLOSED) try: self.connection.close() - except (AttributeError, pika.exceptions.ConnectionWrongStateError): + except pika.exceptions.ConnectionWrongStateError: pass - del self.connection self.callbacks.on_connection_failure(self.name)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rejected/connection.py` around lines 233 - 240, on_failure() currently deletes self.connection which causes later methods like shutdown() to raise AttributeError; remove the del self.connection and instead leave the attribute in place (or set self.connection = None) after attempting close so subsequent code that references self.connection or checks its state won't fail; update the on_failure method (and any related teardown code) to rely on self.connection being present or None rather than deleted, referencing the on_failure() method and shutdown() paths to verify no AttributeError is possible.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rejected/codecs.py`:
- Around line 127-128: The call site currently invokes _load_bs4(body) even when
content_type (from BS4_MIME_TYPES) is XML; update the call to pass content_type
(e.g., _load_bs4(body, content_type)) and change _load_bs4 to accept the
content_type parameter and choose the appropriate parser (use an XML parser for
text/xml vs an HTML parser for text/html, or raise/fallback if the XML parser is
unavailable). Modify references to _load_bs4 and its signature accordingly so
the correct parser is selected based on the passed content_type (and keep
BS4_MIME_TYPES membership logic intact).
- Around line 151-179: The encode() path currently can raise or silently skip
encoding (e.g., _encode_avro → _avro_schema/_load_avro_schema raising
DecodeError, missing umsgpack leaving non-bytes for 'application/msgpack', or
raw exceptions from json.dumps/pickle/plistlib/yaml/gzip/bz2); wrap the entire
serialization and compression block inside a try/except that catches any
Exception and re-raises as EncodeError (preserving the original exception
message), ensure that when content_type == 'application/msgpack' and umsgpack is
None you raise EncodeError (not leave body unencoded), and guarantee the final
body is bytes before applying content_encoding or returning (convert str→bytes
and raise EncodeError if conversion fails); reference encode(), _encode_avro,
_avro_schema/_load_avro_schema, EncodeError, umsgpack, PICKLE_MIME_TYPES,
YAML_MIME_TYPES, BS4_MIME_TYPES in your changes.
In `@rejected/connection.py`:
- Around line 105-111: The try/except around
self.connection.channel(on_open_callback=self.on_channel_open) only catches
pika.exceptions.ConnectionClosed but must also catch
pika.exceptions.ConnectionWrongStateError; update the except clause to handle
ConnectionWrongStateError the same way as ConnectionClosed (log via
LOGGER.warning, call self.set_state(self.STATE_CLOSED), invoke
self.callbacks.on_closed(self.name) and return) so opening a channel when the
connection is in the wrong state is handled consistently; reference the existing
handlers for ConnectionClosed and the later handling at lines 223-224 as the
desired behavior for ConnectionWrongStateError.
- Around line 205-231: The handler currently treats only reply_code <= 0 and 404
as fatal; update the logic in the channel-close handling (the block referencing
reply_code, reply_text, LOGGER, on_failure, is_running, is_shutting_down,
connection.channel, on_channel_open, set_state, STATE_CONNECTING) to also treat
protocol-level channel closure codes 403, 405, and 406 as terminal failures by
calling self.on_failure() (instead of falling through to the reopen branch);
keep existing behavior for shutdown (is_shutting_down) and non-fatal closures,
but ensure any ChannelClosed-like protocol error triggers on_failure
immediately.
---
Duplicate comments:
In `@rejected/connection.py`:
- Around line 233-240: on_failure() currently deletes self.connection which
causes later methods like shutdown() to raise AttributeError; remove the del
self.connection and instead leave the attribute in place (or set self.connection
= None) after attempting close so subsequent code that references
self.connection or checks its state won't fail; update the on_failure method
(and any related teardown code) to rely on self.connection being present or None
rather than deleted, referencing the on_failure() method and shutdown() paths to
verify no AttributeError is possible.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 2bea62cc-e636-4374-bf12-be46b1e54339
📒 Files selected for processing (3)
rejected/codecs.pyrejected/connection.pyrejected/testing.py
🚧 Files skipped from review as they are similar to previous changes (1)
- rejected/testing.py
- Remove pickle deserialization support (RCE vector) - Validate message_type before formatting into schema registry URI - Remove env var dump from Sentry exception reports - Enforce TLS cert verification for non-default SSL protocols - Fix stop_consumer not awaiting async shutdown() coroutine - Fix reject() RuntimeError when no_ack=True during shutdown - Fix STATE_PROCESSING collision with STATE_ACTIVE (0x04 -> 0x09) - Fix GarbageCollectorMixin not reading settings from positional args - Add UV_CONFIG_FILE=/dev/null to CI workflow - Remove S301 ruff suppression (no longer needed without pickle) - Delete stale files: .travis.yml, .codeclimate.yml, README.rst, requires/, init.d/ - Fix redundant import in tests/mocks.py Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Pass content_type to _load_bs4 so XML content uses the xml parser instead of silently applying html.parser to text/xml - Wrap encode() in try/except to maintain EncodeError contract, matching the existing pattern in decode() - Extract _serialize and _compress helpers to reduce encode complexity - Guard against missing umsgpack with explicit EncodeError - Catch ConnectionWrongStateError alongside ConnectionClosed in on_open - Treat AMQP reply codes 403, 405, 406 as terminal channel errors to prevent infinite reopen loops on configuration/permission failures Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Owner
Author
|
🤖 This comment was posted by Claude on behalf of @gmr PR Monitor SummaryCodeRabbit Review Feedback Addressed (commit 87439f7)codecs.py:
connection.py:
All 208 tests pass locally across the changes. Waiting for CI and CodeRabbit re-review on the new commit. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Major restructuring for rejected 4.0 — async Codec class, consolidated pydantic models, consumer hierarchy with backward-compat Consumer and concurrent TransactionConsumer.
Changes
Codec class (
rejected/codecs.py)Codec(schema_registry=None)— asyncencode/decodewith content-type dispatch + AvroSchemaRegistryConfig(type: Literal['file', 'http'], uri: str)replaces per-consumerschema_uri_formathttpxfor async HTTP schema fetching (replacesrequests)asyncio.Lock(double-check pattern)Models (
rejected/models.py)ProcessingContextcarries connection, channel, raw_body, message, measurement, resultMessageis flat (no nested properties/method objects)ResultIntEnum for message dispositionSchemaRegistryConfigwithLiteraltype validationConsumer hierarchy (
rejected/consumer.py)_Consumer: base contract withexecute(ctx)→_pre_execute→_run_consumerConsumer(_Consumer): backward-compat locked single-message,self._contextfor property accessTransactionConsumer(_Consumer): concurrent, receivesProcessingContextinprepare(ctx)/process(ctx)/finish(ctx)publish_messageis async, uses processCodec_republishhelper deduplicates dropped/error republishingraw_bodypreserved for republishing (not decoded body)Process (
rejected/process.py)_in_flightdict tracksProcessingContextper delivery_tagCodecininvoke_consumerType annotations
Other
exceptions.py: standalone modulemeasurement.py: renamed fromdata.pyavro.py: deleted (merged into Codec)httpx>=0.27replacesrequestsin[avro]extraskip-magic-trailing-comma, isort compatTesting
208 tests pass. ruff clean. mypy clean on all modules.
Summary by CodeRabbit
New Features
Documentation
Refactor
Chores