feat(codec): wire codecs into publishing flow by ce1ebrimbor · Pull Request #2902 · ag2ai/faststream

ce1ebrimbor · 2026-06-03T10:16:53Z

Description

Unified codec interface compatible with all the brokers.

WIP #2837

Type of change

Please delete options that are not relevant.

New feature (a non-breaking change that adds functionality)

Checklist

My code adheres to the style guidelines of this project (just lint shows no errors)
I have conducted a self-review of my own code
I have made the necessary changes to the documentation
My changes do not generate any new warnings
I have added tests to validate the effectiveness of my fix or the functionality of my new feature
Both new and existing unit tests pass successfully on my local environment by running just test-coverage
I have ensured that static analysis tests are passing by running just static-analysis
I have included code examples to illustrate the modifications

Lancetnik · 2026-06-03T19:27:43Z

GzipCodec is not prefect example because you can achieve the same behavior using middleware - https://github.com/ulbwa/faststream-compressors

ce1ebrimbor · 2026-06-04T08:22:35Z

GzipCodec is not prefect example because you can achieve the same behavior using middleware - https://github.com/ulbwa/faststream-compressors

I will add a schema registry example, I think this will make more sense.

ce1ebrimbor · 2026-06-04T19:40:12Z

I will add a schema registry example, I think this will make more sense.

@Lancetnik I have added a Schema Registry example I have previously used.
I will need some help for other brokers.

I won't be available next week, if there are new suggestions I will be able to add them after.

ce1ebrimbor · 2026-06-04T19:50:52Z

@Lancetnik I am not comforable with the fact that Rabbit and Redis still need a destination parameter
If we want to keep the consistency we should reworkvthose, your call.
Otherwise let's keep this PR as wip. 🙏

Lancetnik

We should unify behavior and signature for all broker codecs

…and Rabbit producers Unifies the codec API across all brokers. Previously Redis and Rabbit constructed intermediate PublishCommand objects or decomposed the command into scalar arguments before calling codec.encode(). Now the actual RedisPublishCommand/RabbitPublishCommand flows through to the codec, giving it access to destination and all broker-specific metadata. Changes: - redis/parser/message.py: MessageFormat.build() accepts PublishCommand - redis/parser/binary.py: BinaryMessageFormatV1.encode() accepts cmd - redis/publisher/producer.py: pass cmd directly to encode() - rabbit/parser.py: AioPikaParser.encode_message() accepts optional cmd - rabbit/publisher/producer.py: pass RabbitPublishCommand as cmd - rabbit/testing.py: construct RabbitPublishCommand for test messages - Update test callers to use new encode(cmd=...) signature

Move PublishCommand, PublishType imports from function bodies to module level in confluent/publisher/producer.py and confluent/testing.py. Rename _BasePublishCommand to _BaseCmd for consistency.

Move PublishCommand, PublishType imports from function bodies to module level in kafka/publisher/producer.py and kafka/testing.py. Rename _BasePublishCommand to _BaseCmd for consistency.

Move PublishType, PublishCommand imports from function bodies to module level in redis/publisher/producer.py, redis/testing.py, rabbit/parser.py, and rabbit/testing.py.

Since cmd is now required, destination was dead code. Also remove unused PublishType and _BasePublishCommand imports from rabbit/parser.py.

…gnature

Introduce EncodedMessage(body, content_type) dataclass as the return type for CodecProto.encode() and BatchCodecProto.encode_batch(). This replaces the previous tuple[bytes, str | None] contract with a named, extensible structure. All broker producers, testing files, and the schema registry example are updated to use encoded.body and encoded.content_type directly.

…itPublishCommand

… inherits CodecProto

Add warn_deprecated_param() to _compat.py. Emit warnings at: - BrokerConfig.__post_init__ (broker-level parser/decoder) - SubscriberUsecase.add_call() (subscriber-level parser/decoder) - SubscriberUsecase.__call__() (handler-level parser/decoder) Users should migrate to codec with custom encode()/decode().

parser remains valid (raw msg → StreamMessage). Only decoder is replaced by codec.decode(). Inline warnings directly instead of using warn_deprecated_param helper.

ce1ebrimbor · 2026-08-19T08:08:21Z

@Lancetnik if you have time 🙏

Lancetnik · 2026-08-27T11:35:37Z

@ce1ebrimbor hi! Sorry for waiting so long. I'll review and merge the PR this week
Incredible job, thank you!
BTW, can we contact some way? Mail / Linkedin / Telegram / WA? What format works for you?

ce1ebrimbor · 2026-08-27T14:52:03Z

@ce1ebrimbor hi! Sorry for waiting so long. I'll review and merge the PR this week
Incredible job, thank you!
BTW, can we contact some way? Mail / Linkedin / Telegram / WA? What format works for you?

Hey, I will ping you on telegram ^^

ce1ebrimbor requested review from Lancetnik and borisalekseev as code owners June 3, 2026 10:16

github-actions Bot added Confluent Issues related to `faststream.confluent` module AioKafka Issues related to `faststream.kafka` module NATS Issues related to `faststream.nats` module and NATS broker features MQTT Issues related to `faststream.mqtt` module labels Jun 3, 2026

ce1ebrimbor force-pushed the feat/codec-destination branch from d20c610 to 6a33f03 Compare June 3, 2026 10:23

github-actions Bot added the documentation Improvements or additions to documentation label Jun 3, 2026

Lancetnik reviewed Jun 3, 2026

View reviewed changes

Comment thread faststream/confluent/publisher/producer.py Outdated

ce1ebrimbor force-pushed the feat/codec-destination branch from 6a33f03 to 3e0d1d1 Compare June 4, 2026 19:14

github-actions Bot added the Redis Issues related to `faststream.redis` module and Redis features label Jun 4, 2026

ce1ebrimbor force-pushed the feat/codec-destination branch 4 times, most recently from 9b45655 to 75cc013 Compare June 4, 2026 19:38

ce1ebrimbor force-pushed the feat/codec-destination branch 2 times, most recently from 9a14d12 to 1239f71 Compare June 4, 2026 19:45

Lancetnik reviewed Jun 6, 2026

View reviewed changes

Comment thread faststream/confluent/publisher/producer.py Outdated

Lancetnik reviewed Jun 6, 2026

View reviewed changes

Comment thread faststream/confluent/testing.py Outdated

Lancetnik reviewed Jun 6, 2026

View reviewed changes

Comment thread faststream/confluent/testing.py Outdated

Lancetnik reviewed Jun 6, 2026

View reviewed changes

Comment thread faststream/kafka/testing.py Outdated

Lancetnik requested changes Jun 6, 2026

View reviewed changes

ApusBerliozi assigned ce1ebrimbor Aug 3, 2026

ce1ebrimbor marked this pull request as draft August 6, 2026 18:16

ce1ebrimbor force-pushed the feat/codec-destination branch 2 times, most recently from 2b2657e to efe2d10 Compare August 6, 2026 19:46

ce1ebrimbor added 26 commits August 18, 2026 18:49

feat(codec): change encode signature to accept PublishCommand

0a13d8a

docs(codec): replace examples with Schema Registry codec

f0fa7e4

refactor(confluent): move function-level imports to module top

29084db

Move PublishCommand, PublishType imports from function bodies to module level in confluent/publisher/producer.py and confluent/testing.py. Rename _BasePublishCommand to _BaseCmd for consistency.

refactor(kafka): move function-level imports to module top

f0cd4b4

Move PublishCommand, PublishType imports from function bodies to module level in kafka/publisher/producer.py and kafka/testing.py. Rename _BasePublishCommand to _BaseCmd for consistency.

refactor(redis,rabbit): move function-level imports to module top

1a9fd66

Move PublishType, PublishCommand imports from function bodies to module level in redis/publisher/producer.py, redis/testing.py, rabbit/parser.py, and rabbit/testing.py.

chore: remove .omo/ from tracking

ab605d4

refactor: rename _BaseCmd to _BasePublishCommand for consistency

5d6dafe

style: fix linting errors

03fa2a5

refactor(rabbit): remove redundant destination param from encode_message

6ff15ca

Since cmd is now required, destination was dead code. Also remove unused PublishType and _BasePublishCommand imports from rabbit/parser.py.

fix(rabbit): make cmd required in _publish to match encode_message si…

454d3a2

…gnature

style: format codec example

ff5b440

refactor: remove _BasePublishCommand alias, use PublishCommand directly

c99254a

fix: clean up suggestions

81f3b93

docs: use mkdocstrings for CodecProto, fix return type description

8006c97

refactor(rabbit): simplify encode_message to extract params from Rabb…

9735038

…itPublishCommand

ci: add _typos.toml to suppress WellFREEzZ false positive

1cd5447

fix: use super().encode reference test batch brokers

5165587

refactor(tests): use super().encode via encode_item in batch codec tests

08907ba

refactor: re-export codec types from faststream.message, DefaultCodec…

1952e35

… inherits CodecProto

refactor: move warn_deprecated_param import to top of usecase.py

d565255

refactor: deprecate only decoder param, not parser

3db8aeb

parser remains valid (raw msg → StreamMessage). Only decoder is replaced by codec.decode(). Inline warnings directly instead of using warn_deprecated_param helper.

refactor: remove unused warn_deprecated_param helper

04993de

fix: linter

df5e22f

ce1ebrimbor force-pushed the feat/codec-destination branch from 2eddff8 to df5e22f Compare August 18, 2026 16:49

Navigation Menu

feat(codec): wire codecs into publishing flow - #2902

feat(codec): wire codecs into publishing flow#2902
ce1ebrimbor wants to merge 28 commits into
ag2ai:mainfrom
ce1ebrimbor:feat/codec-destination

ce1ebrimbor commented Jun 3, 2026 •

edited

Loading

Uh oh!

Lancetnik commented Jun 3, 2026

Uh oh!

Uh oh!

ce1ebrimbor commented Jun 4, 2026

Uh oh!

ce1ebrimbor commented Jun 4, 2026

Uh oh!

ce1ebrimbor commented Jun 4, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Lancetnik left a comment

Uh oh!

ce1ebrimbor commented Aug 19, 2026

Uh oh!

Lancetnik commented Aug 27, 2026

Uh oh!

ce1ebrimbor commented Aug 27, 2026

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

ce1ebrimbor commented Jun 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Type of change

Checklist

Uh oh!

Lancetnik commented Jun 3, 2026

Uh oh!

Uh oh!

ce1ebrimbor commented Jun 4, 2026

Uh oh!

ce1ebrimbor commented Jun 4, 2026

Uh oh!

ce1ebrimbor commented Jun 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Lancetnik left a comment

Choose a reason for hiding this comment

Uh oh!

ce1ebrimbor commented Aug 19, 2026

Uh oh!

Lancetnik commented Aug 27, 2026

Uh oh!

ce1ebrimbor commented Aug 27, 2026

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ce1ebrimbor commented Jun 3, 2026 •

edited

Loading

ce1ebrimbor commented Jun 4, 2026 •

edited

Loading