0.5.0

What's Changed

This is the biggest change since the creation of FastStream. We have completely refactored the entire package, changing the object registration mechanism, message processing pipeline, and application lifecycle. However, you won't even notice it—we've preserved all public APIs from breaking changes. The only feature not compatible with the previous code is the new middleware.

New features:

1. await FastStream.stop() method and StopApplication exception to stop a FastStream worker are added.

2. broker.subscriber() and router.subscriber() functions now return a Subscriber object you can use later.

subscriber = broker.subscriber("test")

@subscriber(filter = lambda msg: msg.content_type == "application/json")
async def handler(msg: dict[str, Any]):
    ...
 
@subscriber()
async def handler(msg: dict[str, Any]):
    ...

This is the preferred syntax for filtering now (the old one will be removed in 0.6.0)

3. The router.publisher() function now returns the correct Publisher object you can use later (after broker startup).

publisher = router.publisher("test")

@router.subscriber("in")
async def handler():
    await publisher.publish("msg")

(Until 0.5.0 you could use it in this way with broker.publisher only)

4. A list of middlewares can be passed to a broker.publisher as well:

broker = Broker(..., middlewares=())

@broker.subscriber(..., middlewares=())
@broker.publisher(..., middlewares=())  # new feature
async def handler():
    ...

5. Broker-level middlewares now affect all ways to publish a message, so you can encode application outgoing messages here.

6. ⚠️ BREAKING CHANGE ⚠️ : both subscriber and publisher middlewares should be async context manager type

async def subscriber_middleware(call_next, msg):
    return await call_next(msg)

async def publisher_middleware(call_next, msg, **kwargs):
    return await call_next(msg, **kwargs)

@broker.subscriber(
    "in",
    middlewares=(subscriber_middleware,),
)
@broker.publisher(
    "out",
    middlewares=(publisher_middleware,),
)
async def handler(msg):
    return msg

Such changes allow you two previously unavailable features:

• suppress any exceptions and pass fall-back message body to publishers, and
• patch any outgoing message headers and other parameters.

Without those features we could not implement Observability Middleware or any similar tool, so it is the job that just had to be done.
7. A better FastAPI compatibility: fastapi.BackgroundTasks and response_class subscriber option are supported.

8. All .pyi files are removed, and explicit docstrings and methods options are added.

9. New subscribers can be registered in runtime (with an already-started broker):

subscriber = broker.subscriber("dynamic")
subscriber(handler_method)
...
broker.setup_subscriber(subscriber)
await subscriber.start()
...
await subscriber.close()

10. faststream[docs] distribution is removed.

• Update Release Notes for 0.4.7 by @faststream-release-notes-updater in #1295
• 1129 - Create a publish command for the CLI by @MRLab12 in #1151
• Chore: packages upgraded by @davorrunje in #1306
• docs: fix typos by @omahs in #1309
• chore: update dependencies by @Lancetnik in #1323
• docs: fix misc by @Lancetnik in #1324
• docs (#1327): correct RMQ exhcanges behavior by @Lancetnik in #1328
• fix: typer 0.12 exclude by @Lancetnik in #1341
• 0.5.0 by @Lancetnik in #1326
  • close #1103
  • close #840
  • fix #690
  • fix #1206
  • fix #1227
  • close #568
  • close #1303
  • close #1287
  • feat #607
• Generate docs and linter fixes by @davorrunje in #1348
• Fix types by @davorrunje in #1349
• chore: update dependencies by @Lancetnik in #1358
• feat: final middlewares by @Lancetnik in #1357
• Docs/0.5.0 features by @Lancetnik in #1360

New Contributors

• @MRLab12 made their first contribution in #1151
• @omahs made their first contribution in #1309

Full Changelog: 0.4.7...0.5.0

v0.5.0rc2

What's Changed

This is the final API change before stable 0.5.0 release

⚠️ HAS BREAKING CHANGE

In it, we stabilize the behavior of publishers & subscribers middlewares

async def subscriber_middleware(call_next, msg):
    return await call_next(msg)

async def publisher_middleware(call_next, msg, **kwargs):
    return await call_next(msg, **kwargs)

@broker.subscriber(
    "in",
    middlewares=(subscriber_middleware,),
)
@broker.publisher(
    "out",
    middlewares=(publisher_middleware,),
)
async def handler(msg):
    return msg

Such changes allows you two features previously unavailable

• suppress any exceptions and pas fall-back message body to publishers
• patch any outgoing message headers and other parameters

Without these features we just can't impelement Observability Middleware or any similar tool, so it is the job to be done.

Now you are free to get access at any message processing stage and we are one step closer to the framework we would like to create!

• Update Release Notes for 0.5.0rc0 by @faststream-release-notes-updater in #1347
• Generate docs and linter fixes by @davorrunje in #1348
• Fix types by @davorrunje in #1349
• chore: update dependencies by @Lancetnik in #1358
• feat: final middlewares by @Lancetnik in #1357

Full Changelog: 0.5.0rc0...0.5.0rc2

v0.5.0rc0

What's Changed

This is the biggest change since the creation of FastStream. We have completely refactored the entire package, changing the object registration mechanism, message processing pipeline, and application lifecycle. However, you won't even notice it—we've preserved all public APIs from breaking changes. The only feature not compatible with the previous code is the new middleware.

This is still an RC (Release Candidate) for you to test before the stable release. You can manually install it in your project:

pip install faststream==0.5.0rc0

We look forward to your feedback!

New features:

1. await FastStream.stop() method and StopApplication exception to stop a FastStream worker are added.

2. broker.subscriber() and router.subscriber() functions now return a Subscriber object you can use later.

subscriber = broker.subscriber("test")

@subscriber(filter = lambda msg: msg.content_type == "application/json")
async def handler(msg: dict[str, Any]):
    ...
 
@subscriber()
async def handler(msg: dict[str, Any]):
    ...

This is the preferred syntax for filtering now (the old one will be removed in 0.6.0)

3. The router.publisher() function now returns the correct Publisher object you can use later (after broker startup).

publisher = router.publisher("test")

@router.subscriber("in")
async def handler():
    await publisher.publish("msg")

(Until 0.5.0 you could use it in this way with broker.publisher only)

4. A list of middlewares can be passed to a broker.publisher as well:

broker = Broker(..., middlewares=())

@broker.subscriber(..., middlewares=())
@broker.publisher(..., middlewares=())  # new feature
async def handler():
    ...

5. Broker-level middlewares now affect all ways to publish a message, so you can encode application outgoing messages here.

6. ⚠️ BREAKING CHANGE ⚠️ : both subscriber and publisher middlewares should be async context manager type

from contextlib import asynccontextmanager

@asynccontextmanager
async def subscriber_middleware(msg_body):
    yield msg_body

@asynccontextmanager
async def publisher_middleware(
    msg_to_publish,
    **publish_arguments,
):
    yield msg_to_publish

@broker.subscriber("in", middlewares=(subscriber_middleware,))
@broker.publisher("out", middlewares=(publisher_middleware,))
async def handler():
    ...

7. A better FastAPI compatibility: fastapi.BackgroundTasks and response_class subscriber option are supported.

8. All .pyi files are removed, and explicit docstrings and methods options are added.

9. New subscribers can be registered in runtime (with an already-started broker):

subscriber = broker.subscriber("dynamic")
subscriber(handler_method)
...
broker.setup_subscriber(subscriber)
await subscriber.start()
...
await subscriber.close()

10. faststream[docs] distribution is removed.

• Update Release Notes for 0.4.7 by @faststream-release-notes-updater in #1295
• 1129 - Create a publish command for the CLI by @MRLab12 in #1151
• Chore: packages upgraded by @davorrunje in #1306
• docs: fix typos by @omahs in #1309
• chore: update dependencies by @Lancetnik in #1323
• docs: fix misc by @Lancetnik in #1324
• docs (#1327): correct RMQ exhcanges behavior by @Lancetnik in #1328
• fix: typer 0.12 exclude by @Lancetnik in #1341
• 0.5.0 by @Lancetnik in #1326
• close #1103
• close #840
• fix #690
• fix #1206
• fix #1227
• close #568
• close #1303
• close #1287
• feat #607

New Contributors

• @MRLab12 made their first contribution in #1151
• @omahs made their first contribution in #1309

Full Changelog: 0.4.7...0.5.0rc0

v0.4.7

What's Changed

• Update Release Notes for 0.4.6 by @faststream-release-notes-updater in #1286
• fix (#1263): correct nested descriminator msg type AsyncAPI schema by @Lancetnik in #1288
• docs: add apply_types warning notice to subscription/index.md by @Lancetnik in #1291
• chore: fixed nats-py version by @Lancetnik in #1294

Full Changelog: 0.4.6...0.4.7

v0.4.6

What's Changed

• Add poll in confluent producer to fix BufferError by @kumaranvpl in #1277
• Cover confluent asyncapi tests by @kumaranvpl in #1279
• chore: bump package versions by @davorrunje in #1285

Full Changelog: 0.4.5...0.4.6

v0.4.5

What's Changed

• Update Release Notes for 0.4.4 by @faststream-release-notes-updater in #1260
• Removed unused pytest dependecy from redis/schemas.py by @ashambalev in #1261
• chore: bumped package versions by @davorrunje in #1270
• fix (#1263): correct AsyncAPI schema in descriminator case by @Lancetnik in #1272

New Contributors

• @ashambalev made their first contribution in #1261

Full Changelog: 0.4.4...0.4.5

v0.4.4

What's Changed

Add RedisStream batch size option

@broker.subscriber(stream=StreamSub("input", batch=True, max_records=3))
async def on_input_data(msgs: list[str]):
    assert len(msgs) <= 3

• Update Release Notes for 0.4.3 by @faststream-release-notes-updater in #1247
• docs: add manual run section by @Lancetnik in #1249
• feat (#1252): respect Redis StreamSub last_id with consumer group by @Lancetnik in #1256
• fix: correct Redis consumer group behavior by @Lancetnik in #1258
• feat: add Redis Stream max_records option by @Lancetnik in #1259

Full Changelog: 0.4.3...0.4.4

v0.4.3

What's Changed

Allow to specify Redis Stream maxlen option in publisher:

@broker.publisher(stream=StreamSub("Output", maxlen=10))
async def on_input_data():
    ....

• chore: bump version by @Lancetnik in #1198
• Update Release Notes for 0.4.2 by @faststream-release-notes-updater in #1199
• Add missing API documentation for apply_pattern by @kumaranvpl in #1201
• chore: polishing by @davorrunje in #1203
• Comment out retry and timeout in a confluent test by @kumaranvpl in #1207
• Commit offsets only if auto_commit is True by @kumaranvpl in #1208
• Add a CI job to check for missed docs changes by @kumaranvpl in #1217
• fix: inconsistent NATS publisher signature by @Lancetnik in #1218
• Upgrade packages by @davorrunje in #1226
• chore: bump dawidd6/action-download-artifact from 3.0.0 to 3.1.1 by @dependabot in #1239
• chore: bump dependencies by @Lancetnik in #1246
• feat (#1235): StreamSub maxlen parameter by @Lancetnik in #1245
• fix (#1234): correct FastAPI path passing, fix typehints by @Lancetnik in #1236
• fix (#1231): close RMQ while reconnecting by @Lancetnik in #1238

Full Changelog: 0.4.2...0.4.3

v0.4.2

What's Changed

Bug fixes

• fix: correct RMQ Topic testing routing by @Lancetnik in #1196
• fix #1191: correct RMQ ssl default port by @Lancetnik in #1195
• fix #1143: ignore Depends in AsyncAPI by @Lancetnik in #1197

Full Changelog: 0.4.1...0.4.2

v0.4.1

What's Changed

Bug fixes

• Fix: use FastAPI overrides in subscribers by @Lancetnik in #1189
• Handle confluent consumer commit failure by @kumaranvpl in #1193

Documentation

• Include Confluent in home and features pages by @kumaranvpl in #1186
• Use pydantic model for publishing in docs example by @kumaranvpl in #1187

Full Changelog: 0.4.0...0.4.1