v0.6.0rc0

Description

FastStream 0.6 is a significant technical release that aimed to address many of the current project design issues and unlock further improvements on the path to version 1.0.0. We tried our best to minimize breaking changes, but unfortunately, some aspects were simply not working well. Therefore, we decided to break them in order to move forward.

This release includes:

• Finalized Middleware API
• Finalized Router API
• Introduced dynamic subscribers
• Added support for various serializer backends (such as Msgspec)
• Support for AsyncAPI 3.0 specification
• A range of minor refactors and improvements

The primary goal of this release is to unlock the path towards further features. Therefore, we are pleased to announce that after this release, we plan to work on MQTT #956 and SQS #794 support and move towards version 1.0.0!

Breaking changes

Firstly, we have dropped support for Python 3.8 and Python 3.9. Python 3.9 is almost at the end of its life cycle, so it's a good time to update our minimum version.

FastStream object changes

The broker has become a POSITIONAL-ONLY argument. This means that FastStream(broker=broker) is no longer valid. You should always pass the broker as a separate positional argument, like FastStream(brokers), to ensure proper usage.

This is a preparatory step for FastStream(*brokers) support, which will be introduced in 1.0.0.

AsyncAPI changes

In 0.6, you can't directly pass custom AsyncAPI options to the FastStream constructor anymore.

app = FastStream(   # doesn't work anymore
    ...,
    title="My App",
    version="1.0.0",
    description="Some desctiption",
)

You need to create a specification object and pass it manually to the constructor.

from faststream import FastStream, AsyncAPI

FastStream(
    ...
    specification=AsyncAPI(
        title="My App",
        version="1.0.0",
        description="Some desctiption",
    )
)

Retry feature removed

Previously, you were able to configure retry attempts for a handler by using the following option:

@broker.subscriber("in", retry=True)  # was removed
async def handler(): ...

Unfortunately, this option was a design mistake. We apologize for any confusion it may have caused. Technically, it was just a shortcut to message.nack() on error. We have decided that manual acknowledgement control would be more idiomatic and better for the framework. Therefore, we have provided a new feature in its place: ack_policy control.

@broker.subscriber("test", ack_policy=AckPolicy.ACK_FIRST)
async def handler() -> None: ...

With ack_policy, you can now control the default acknowledge behavior for your handlers. AckPolicy offers the following options:

• REJECT_ON_ERROR (default) – to permanently discard messages on failure.
• NACK_ON_ERROR – to redeliver messages in case of failure.
• ACK_FIRST – for scenarios with high throughput where some message loss can be acceptable.
• ACK – if you want the message to be acknowledged, regardless of success or failure.
• MANUAL – fully manually control message acknowledgment (for example, calling #!python message.ack() yourself).

In addition, we have deprecated a few more options prior to ack_policy.

• ack_first=True -> AckPolicy.ACK_FIRST
• no_ack=True -> AckPolicy.MANUAL

Context changes

We have made some changes to our Dependency Injection system, so the global context is no longer available.

Currently, you cannot simply import the context from anywhere and use it freely.

from faststeam import context  # was removed

Instead, you should create the context in a slightly different way. The FastStream object serves as an entry point for this, so you can place it wherever you need it:

from typing import Annotated

from faststream import Context, ContextRepo, FastStream
from faststream.rabbit import RabbitBroker

broker = RabbitBroker()

app = FastStream(
    broker,
    context=ContextRepo({
        "global_dependency": "value",
    }),
)

Everything else about using the context remains the same. You can request it from the context at any place that supports it.

Additionally, Context("broker") and Context("logger") have been moved to the local context. They cannot be accessed from lifespan hooks any longer.

@app.after_startup
async def start(
    broker: Broker   # does not work anymore
): ...

@router.subscriber
async def handler(
    broker: Broker   # still working
): ...

This change was also made to support multiple brokers.

Middlewares changes

Also, we have finalized our Middleware API. It now supports all the features we wanted, and we have no plans to change it anymore. First of all, the BaseMiddleware class constructor requires a context (which is no longer global).

class BaseMiddleware:
    def __init__(self, msg: Any | None, context: ContextRepo) -> None:
        self.msg = msg
        self.context = context

The context is now available as self.context in all middleware methods.

We also changed the publish_scope function signature.

class BaseMiddleware:   # old signature
    async def publish_scope(
        self,
        call_next: "AsyncFunc",
        msg: Any,
        *args: Any,
        **kwargs: Any,
    ) -> Any: ...

Previously, any options passed to brocker.publish("msg", "destination") had to be consumed as *args, **kwargs.

Now, you can consume them all as a single PublishCommand object.

from faststream import PublishCommand

class BaseMiddleware:
    async def publish_scope(
        self,
        call_next: Callable[[PublishCommand], Awaitable[Any]],
        cmd: PublishCommand,
    ) -> Any: ...

Thanks to Python 3.13's TypeVars with defaults, BaseMiddleware becomes a generic class and you can specify the PublishCommand for the broker you want to work with.

from faststream.rabbit import RabbitPublishCommand

class Middleware(BaseMiddleware[RabbitPublishCommand]):
    async def publish_scope(
        self,
        call_next: Callable[[RabbitPublishCommand], Awaitable[Any]],
        cmd: RabbitPublishCommand,
    ) -> Any: ...

Warning: The methods on_consume, after_consume, on_publish and after_publish will be deprecated and removed in version 0.7. Please use consume_scope and publish_scope instead.

Redis Default Message format changes

In FastStream 0.6 we are using BinaryMessageFormatV1 as a default instead of JSONMessageFormat .
You can find more details in the documentation: https://faststream.ag2.ai/latest/redis/message_format/

New Features:

1. AsyncAPI3.0 support – now you can choose between AsyncAPI(schema_version="3.0.0") (default) and AsyncAPI(schema_version="2.6.0") schemas generation

2. Msgspec native support

   from fast_depends.msgspec import MsgSpecSerializer
   
   broker = Broker(serializer=MsgSpecSerializer())

3. Subscriber iteration support. This features supports all middlewares and other FastStream features.

   subscriber = broker.subscriber(...)
   
   await subscriber.start()
   
   async for msg in subscriber:
       ...

Deprecation removed

1. @broker.subscriber(..., filters=...) removed
2. message.decoded_body removed, use await message.decode() instead
3. publish(..., rpc=True) removed, use broker.request() instead
4. RabbitMQ @broker.subscriber(..., reply_config=...) removed, use Response instead

Related Issues

1. fixes #1742
2. close #1228
3. close #980
4. fixes #1742
5. feature #1895
6. fixes #1954
7. close #1646
8. fixes #1625
9. close #1904
10. close #1507
11. close #2056
12. close #1308
13. close #1901
14. close #2029
15. close #2094
16. close #1881
17. close #2216
18. close #2215
19. close #2031
20. close #2178
21. close #2239
22. fixes #1036

New Contributors

• @stepanbobrik made their first contribution in #2381

Full Changelog: 0.5.48...0.6.0rc0

v0.5.48

What's Changed

This release is part of the migration to FastStream 0.6.

In order to provide great features such as observability and more, FastStream requires the inclusion of additional data in your messages. Redis, on the other hand, allows for the sending of any type of data within a message. Therefore, with this release, we introduce FastStream's own binary message format, which supports any data type you wish to use and can include additional information.

For more information on the message format, please see the documentation

By default, we are still using the JSON message format, but as of version 0.6, the default will change to the binary format. Therefore, you can prepare your services for this change by manually setting a new protocol.

For whole broker:

from faststream.redis import RedisBroker, BinaryMessageFormatV1

# JSONMessageFormat using by default, but it will be deprecated in future updates
broker = RedisBroker(message_format=BinaryMessageFormatV1)

Or for a specifica subscriber / publisher

from faststream.redis import RedisBroker, BinaryMessageFormatV1

broker = RedisBroker()

@broker.subscriber(..., message_format=BinaryMessageFormatV1)
@broker.publisher(..., message_format=BinaryMessageFormatV1)
async def handler(msg):
    return msg

Special thanks for @ilya-4real for this great feature!

FastStream require a li

• docs: detach index pages from sections by @Lancetnik in #2358
• chore: Force push gh-pages by @dolfinus in #2360
• fix: #2365 respect prefix by publisher in include_router case by @kodsurfer in #2366
• feature: RMQ reply-to explicit exchange by @vesalius512 in #2368
• Added the ability to specify any start identifier for the xreadgroup by @powersemmi in #2309
• feat: (#2061) Redis Binary Format by @ilya-4real in #2287

New Contributors

• @kodsurfer made their first contribution in #2366
• @vesalius512 made their first contribution in #2368
• @powersemmi made their first contribution in #2309

Full Changelog: 0.5.47...0.5.48

v0.5.47

What's Changed

• fix: correct NATS pattern AsyncAPI render by @Lancetnik in #2354

Full Changelog: 0.5.46...0.5.47

v0.5.46

What's Changed

• fix overlapping patterns by @xgemx in #2349

New Contributors

• @xgemx made their first contribution in #2349

Full Changelog: 0.5.45...0.5.46

v0.5.45

What's Changed

• ci: polish workflows by @Lancetnik in #2331
• Update Release Notes for main by @davorrunje in #2337
• Markdown Code Blocks Auto Set Language by @mahenzon in #2339
• Change docs by @Flosckow in #2343
• chore(deps): bump the pip group with 5 updates by @dependabot in #2344
• ci: fix jobs by @Lancetnik in #2338

Full Changelog: 0.5.44...0.5.45

0.5.44

What's Changed

• Cli tests by @borisalekseev in #2295
• feature #2091: AsyncAPI HTTP support by @aligeromachine in #2301
• add try/finally to ensure we exit on close error by @JonathanSerafini in #2304
• docs: fix EmailStr duplication in API Reference by @AliakseiYafremau in #2308
• Rewrite AsyncAPI docs and log file config tests by @borisalekseev in #2302
• Add logging kafka errors by @spataphore1337 in #2318
• Add aiokafka error handling by @spataphore1337 in #2319
• feat: add a manual action for deploying from gh-pages branch by @Tapeline in #2325
• Gh pages manual trigger fix by @Tapeline in #2326
• feat: add CODEOWNERS by @draincoder in #2330
• introduce stop method on broker and subscriber and deprecate close by @mahenzon in #2328
• fix: add injection extra options in lifespan by @Sehat1137 in #2335

New Contributors

• @aligeromachine made their first contribution in #2301
• @AliakseiYafremau made their first contribution in #2308
• @mahenzon made their first contribution in #2328

Full Changelog: 0.5.43...0.5.44

v0.5.43

What's Changed

• feat (#2291): add global_qos option to Channel by @Lancetnik in #2297
• feat: added delete method to redis message by @ilya-4real in #2294
• feat: add CLI extra options by @Sehat1137 in #2231
• feat: redis pipeline by @Maclovi in #2270
• fix: replace passive with declare by @Maclovi in #2236
• fix: deprecate log fmt by @Maclovi in #2240
• fix (#2252): remove useless code snippet by @Lancetnik in #2256
• fix (depends): removed SetupError if correct fastapi.Depends is included to annotation by @NelsonNotes in #2280
• Improve license badge by @sobolevn in #2241
• chore: use stricter mypy settings by @sobolevn in #2243
• docs: alwayse use ' in dependency groupds by @sobolevn in #2242
• docs: better highlight first example by @sobolevn in #2245
• docs: add more links to readme by @sobolevn in #2246
• docs: Improve README code examples by @sobolevn in #2247
• docs: reword Multiple Subscriptions section by @sobolevn in #2250
• docs: Improve README code examples by @mxnoob in #2253
• docs: highlight more code by @sobolevn in #2255
• docs: Fix code style in Context Fields Declaration by @sobolevn in #2269
• docs: Improve Context docs by @sobolevn in #2265
• docs: Reword publishing/decorator.md by @sobolevn in #2260
• docs: Imporve dependencies docs by @sobolevn in #2263
• docs: Refactor publishing/index.md docs by @sobolevn in #2259
• docs: Improve wordings in test docs by @sobolevn in #2258
• docs: Link to test.md from initial tutorial by @sobolevn in #2257
• docs: Improve the documentation page with Response class by @anywindblows in #2238
• docs: fix broken links by @Lancetnik in #2271
• docs: Fix serialization/index.md wording and syntax by @sobolevn in #2277
• docs: import types from correct place by @sobolevn in #2278
• docs: Improve serialization examples by @sobolevn in #2282
• docs: improve Lifespan: Hooks page by @sobolevn in #2284
• docs: -1/+1 fix mistake by @Sehat1137 in #2286
• docs: doc fix by @RenameMe1 in #2288
• docs: remove deprecated syntax from examples by @Lancetnik in #2234
• docs: Improve grammar in Lifespan: testing by @sobolevn in #2298

New Contributors

• @mxnoob made their first contribution in #2253
• @anywindblows made their first contribution in #2238
• @ilya-4real made their first contribution in #2294

Full Changelog: 0.5.42...0.5.43

v0.5.42

What's Changed

• Feature: add deprecate on retry arg by @Flosckow in #2224
• Fix default rabbit timestamp by @dmder in #2226
• Fix ASGIMultiprocess args mismatch by @dolfinus in #2228
• Add docs about cli --log-file by @RenameMe1 in #2229
• fix (#2227): use ms as generated timestamp in Kafka by @Lancetnik in #2230
• feat: Add AsyncAPI HTTP Support by @tmulligan98 in #2142

New Contributors

• @dmder made their first contribution in #2226
• @tmulligan98 made their first contribution in #2142

Full Changelog: 0.5.41...0.5.42

v0.5.41

What's Changed

• feature: injection FastAPI in StreamMessage scope by @IvanKirpichnikov in #2205
• feat: logging configuration from file by @RenameMe1 in #2203
• feat: add cluster metadata request via aiokafka admin client in ping by @murzinov01 in #2212

New Contributors

• @murzinov01 made their first contribution in #2212

Full Changelog: 0.5.40...0.5.41

v0.5.40

What's Changed

• fix: repair AsyncAPI render by @Sehat1137 in #2175
• fix: prevent redis stream subscriber timeout when polling_interval exceeds 3000ms by @caozheliang in #2200
• feat: auto flush confluent broker by @mdaffad in #2182
• feat (#2169): Error on faststream.Context instead of faststream.[broker].fastapi.Context by @NelsonNotes in #2181
• docs: add information about manual AsyncAPI hosting by @Sehat1137 in #2177
• docs: update Django integration guide and add FastStream ORM access e… by @Lodimup in #2187
• docs: add Django management command integration by @Lodimup in #2190
• chore: bump redis to 6.0.0 by @Lancetnik in #2201
• chore: deprecate connect options by @Lancetnik in #2202

New Contributors

• @Lodimup made their first contribution in #2187
• @caozheliang made their first contribution in #2200

Full Changelog: 0.5.39...0.5.40