Feature: Allow broker creation in on_startup hook

[antoinehumbert]

For some reason, I need to delay the creation of an app's broker in an on_startup hook. This is mainly because my application has a pretty heavy warmup bootstrap that loads a lot of data into memory so I need to delay this bootstrap (which actually includes broker creation) after the parent process forked into worker subprocesses, otherwise the parent process has an unnecessary high memory footprint (that only worker processes need to have).

E.g.

from faststream import FastStream
from faststream.kafka import KafkaBroker

app = FastStream()

@app.on_startup
def bootstrap():
    app.set_broker(KafkaBroker(["localhost:9092"]))
    # some warmup of the application pre-loading data into memory
    ...

For now, this isn't working because of this assert

faststream/faststream/app.py

Line 40 in 9b7c33e

assert self.broker, "You should setup a broker" # nosec B101

which occurs at the very begining of the workers run, before on_startup hooks are executed.

The example above is overly simplified and one could argue that broker could have been created outside the bootstrap() function, but, as per example at https://faststream.airt.ai/latest/getting-started/lifespan/hooks/#lifespan, it could be legitimate to create it based on environment setting specifying not only the broker host but also the kind of broker to use (nats, kafka, ...).

I currently solve the problem by providing an empty KafkaBroker at app instanciation (app = FastStream(broker=KafkaBroker())) before overriding it using app.set_broker() in the bootstrap() function.

However, as per the base Application constructor at

faststream/faststream/_internal/application.py

Lines 48 to 70 in 9b7c33e

	class Application(ABC, AsyncAPIApplication):
	def __init__(
	self,
	broker: Optional["BrokerUsecase[Any, Any]"] = None,
	logger: Optional["LoggerProto"] = logger,
	lifespan: Optional["Lifespan"] = None,
	# AsyncAPI args,
	title: str = "FastStream",
	version: str = "0.1.0",
	description: str = "",
	terms_of_service: Optional["AnyHttpUrl"] = None,
	license: Optional[Union["License", "LicenseDict", "AnyDict"]] = None,
	contact: Optional[Union["Contact", "ContactDict", "AnyDict"]] = None,
	tags: Optional[Sequence[Union["Tag", "TagDict", "AnyDict"]]] = None,
	external_docs: Optional[
	Union["ExternalDocs", "ExternalDocsDict", "AnyDict"]
	] = None,
	identifier: Optional[str] = None,
	on_startup: Sequence[Callable[P_HookParams, T_HookReturn]] = (),
	after_startup: Sequence[Callable[P_HookParams, T_HookReturn]] = (),
	on_shutdown: Sequence[Callable[P_HookParams, T_HookReturn]] = (),
	after_shutdown: Sequence[Callable[P_HookParams, T_HookReturn]] = (),
	) -> None:

which allows application instanciation without a defined broker and this comment in Application.set_broker()

faststream/faststream/_internal/application.py

Lines 117 to 120 in 9b7c33e

	def set_broker(self, broker: "BrokerUsecase[Any, Any]") -> None:
	"""Set already existed App object broker.
	Useful then you create/init broker in `on_startup` hook.

which claims to be used for creation/init in on_startup, I would have expected my use case to be legitimate.

As a consequence, should this assert be removed, moved after startup hooks execution (at

faststream/faststream/_internal/application.py

Lines 168 to 175 in 9b7c33e

	async def start(
	self,
	**run_extra_options: "SettingField",
	) -> None:
	"""Executes startup hooks and start broker."""
	for func in self._on_startup_calling:
	await func(**run_extra_options)

) or just replaced by a warning when broker is None at

faststream/faststream/_internal/application.py

Lines 176 to 177 in 9b7c33e

	if self.broker is not None:
	await self.broker.start()