feat: merge `Broker(context=...)` and `FastStream(context=...)` at broker-level

[nectarindev]

I recently migrated my project to faststream 0.6. I was very interested in how I could add my dependencies to the context. Prior to version 0.6, I did something like this:

from faststream.annotations import ContextRepo
from faststream.kafka import KafkaBroker
from faststream.utils.context import context

broker = KafkaBroker()


@broker.subscriber("my_topic", group_id="my_group")
async def handle(
    context: ContextRepo,
):
    print("dependency: ", context.get("dependency"))  # 42


async def lifespan(*args, **kwargs):
    context.set_global("dependency", 42)

    await broker.start()
    try:
        yield
    finally:
        await broker.stop()

I launched the broker as part of my application in lifespan without using the FastStream class.

For version 0.6, I saw examples where it was suggested to pass the context to FastStream, but that solution did not suit me. I discovered that the broker also accepts context, and that solves my problem:

broker = KafkaBroker(context=ContextRepo({"dependency": 42}))

...

async def lifespan(*args, **kwargs):
    await broker.start()
    try:
        yield
    finally:
        await broker.stop()

But I also discovered that if I create a FastStream instance, its context will be used, even though I didn't use it to start the broker.

from fastapi import FastAPI
from faststream import ContextRepo, FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker(context=ContextRepo({"broker_dependency": 2}))
app = FastStream(broker, context=ContextRepo({"application_dependency": 1}))


@broker.subscriber("my_topic", group_id="my_group")
async def handle(
    context: ContextRepo,
):
    print("broker_dependency: ", context.get("broker_dependency"))            # None
    print("application_dependency: ", context.get("application_dependency"))  # 1


async def lifespan(*args, **kwargs):
    await broker.start()
    try:
        yield
    finally:
        await broker.stop()


asgi = FastAPI(lifespan=lifespan)

I'm not sure that's normal behavior. It would make much more sense if only the broker's dependency were available.

---

Running FastStream 0.6.3 with CPython 3.12.4 on Linux

[nectarindev]

I also have some thoughts that are no longer related to the error.

When running via faststream run cli, we get the same result, although we could merge the broker and FastStream contexts.

In general, in a future multi-broker environment, we could do something like this:

[Lancetnik]

@nectarindev yes, such design was made with multibrokers support in a head. So, the major case is FastStream-level context. We added Broker-level option just in case, when you want to use broker without application object

But, it should be easy to fix:

• https://github.com/ag2ai/faststream/blob/main/faststream/_internal/broker/broker.py#L88 - here we merge two configs
• https://github.com/ag2ai/faststream/blob/main/faststream/_internal/configs/broker.py#L62 - in this composition we use just the outer config (application one), but, we can redesign it to merge variables instead of overrides

But, I think, we should just add a warning (or even error) about broker-level config overriding by application one to prevent such unexpected usages

[nectarindev]

I think it's worth merging the configs. This will allow you to set the isolated contexts in brokers without fear of it being overwritten by the outer config in scenarios where the user application supports a hybrid launch method, for example, as part of an ASGI application, or only brokers via FastStream.