Skip to content

FastStream

faststream.FastStream #

FastStream(
    broker: Optional[BrokerUsecase[Any, Any]] = None,
    /,
    logger: Optional[LoggerProto] = logger,
    provider: Optional[Provider] = None,
    serializer: Optional[SerializerProto] = EMPTY,
    context: ContextRepo | None = None,
    lifespan: Optional[Lifespan] = None,
    on_startup: Sequence[AnyCallable] = (),
    after_startup: Sequence[AnyCallable] = (),
    on_shutdown: Sequence[AnyCallable] = (),
    after_shutdown: Sequence[AnyCallable] = (),
    specification: Optional[SpecificationFactory] = None,
)

Bases: Application

A class representing a FastStream application.

Source code in faststream/app.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    broker: Optional["BrokerUsecase[Any, Any]"] = None,
    /,
    logger: Optional["LoggerProto"] = logger,
    provider: Optional["Provider"] = None,
    serializer: Optional["SerializerProto"] = EMPTY,
    context: ContextRepo | None = None,
    lifespan: Optional["Lifespan"] = None,
    on_startup: Sequence["AnyCallable"] = (),
    after_startup: Sequence["AnyCallable"] = (),
    on_shutdown: Sequence["AnyCallable"] = (),
    after_shutdown: Sequence["AnyCallable"] = (),
    specification: Optional["SpecificationFactory"] = None,
) -> None:
    super().__init__(
        broker,
        logger=logger,
        config=FastDependsConfig(
            provider=provider or dependency_provider,
            context=context or ContextRepo(),
            serializer=serializer,
        ),
        lifespan=lifespan,
        on_startup=on_startup,
        after_startup=after_startup,
        on_shutdown=on_shutdown,
        after_shutdown=after_shutdown,
        specification=specification,
    )

    self._should_exit = False

context property #

context: ContextRepo

broker property #

broker: Optional[BrokerUsecase[Any, Any]]

logger instance-attribute #

logger = logger

lifespan_context instance-attribute #

lifespan_context = apply_types(
    func=lifespan,
    serializer_cls=_serializer,
    cast_result=False,
    context__=context,
)

set_broker #

set_broker(broker: BrokerUsecase[Any, Any]) -> None

Set already existed App object broker.

Useful then you create/init broker in on_startup hook.

Source code in faststream/_internal/application.py 
106
107
108
109
110
111
112
113
114
115
116
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.
    """
    if self.brokers:
        msg = f"`{self}` already has a broker. You can't use multiple brokers until 1.0.0 release."
        raise SetupError(msg)

    self.brokers.append(broker)
    _ = self.schema.add_broker(broker)

start async #

start(**run_extra_options: SettingField) -> None

Executes startup hooks and start broker.

Source code in faststream/_internal/application.py 
205
206
207
208
209
210
211
async def start(
    self,
    **run_extra_options: "SettingField",
) -> None:
    """Executes startup hooks and start broker."""
    async with self._start_hooks_context(**run_extra_options):
        await self._start_broker()

stop async #

stop() -> None

Executes shutdown hooks and stop broker.

Source code in faststream/_internal/application.py 
254
255
256
257
258
async def stop(self) -> None:
    """Executes shutdown hooks and stop broker."""
    async with self._shutdown_hooks_context():
        for broker in self.brokers:
            await broker.stop()

on_startup #

on_startup(
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]

Add hook running BEFORE broker connected.

This hook also takes an extra CLI options as a kwargs.

Source code in faststream/_internal/application.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def on_startup(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running BEFORE broker connected.

    This hook also takes an extra CLI options as a kwargs.
    """
    self._on_startup_calling.append(
        apply_types(
            to_async(func),
            serializer_cls=self.config._serializer,
            context__=self.context,
        ),
    )
    return func

on_shutdown #

on_shutdown(
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]

Add hook running BEFORE broker disconnected.

Source code in faststream/_internal/application.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
def on_shutdown(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running BEFORE broker disconnected."""
    self._on_shutdown_calling.append(
        apply_types(
            to_async(func),
            serializer_cls=self.config._serializer,
            context__=self.context,
        ),
    )
    return func

after_startup #

after_startup(
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]

Add hook running AFTER broker connected.

Source code in faststream/_internal/application.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
def after_startup(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running AFTER broker connected."""
    self._after_startup_calling.append(
        apply_types(
            to_async(func),
            serializer_cls=self.config._serializer,
            context__=self.context,
        ),
    )
    return func

after_shutdown #

after_shutdown(
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]

Add hook running AFTER broker disconnected.

Source code in faststream/_internal/application.py 
335
336
337
338
339
340
341
342
343
344
345
346
347
def after_shutdown(
    self,
    func: Callable[P_HookParams, T_HookReturn],
) -> Callable[P_HookParams, T_HookReturn]:
    """Add hook running AFTER broker disconnected."""
    self._after_shutdown_calling.append(
        apply_types(
            to_async(func),
            serializer_cls=self.config._serializer,
            context__=self.context,
        ),
    )
    return func

run async #

run(
    log_level: int = INFO,
    run_extra_options: dict[str, SettingField]
    | None = None,
    sleep_time: float = 0.1,
) -> None

Run FastStream Application.

Source code in faststream/app.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def run(
    self,
    log_level: int = logging.INFO,
    run_extra_options: dict[str, "SettingField"] | None = None,
    sleep_time: float = 0.1,
) -> None:
    """Run FastStream Application."""
    set_exit(lambda *_: self.exit(), sync=False)

    async with self.lifespan_context(**(run_extra_options or {})):
        try:
            async with anyio.create_task_group() as tg:
                tg.start_soon(self._startup, log_level, run_extra_options)

                while not self._should_exit:  # noqa: ASYNC110 (requested by creator)
                    await anyio.sleep(sleep_time)

                await self._shutdown(log_level)
                tg.cancel_scope.cancel()
        except ExceptionGroup as e:
            for ex in e.exceptions:
                raise ex from None

exit #

exit() -> None

Stop application manually.

Source code in faststream/app.py 
102
103
104
def exit(self) -> None:
    """Stop application manually."""
    self._should_exit = True

as_asgi #

as_asgi(
    asgi_routes: Sequence[tuple[str, ASGIApp]] = (),
    asyncapi_path: Union[str, AsyncAPIRoute, None] = None,
) -> AsgiFastStream
Source code in faststream/app.py 
106
107
108
109
110
111
112
113
114
115
def as_asgi(
    self,
    asgi_routes: Sequence[tuple[str, "ASGIApp"]] = (),
    asyncapi_path: Union[str, "AsyncAPIRoute", None] = None,
) -> AsgiFastStream:
    return AsgiFastStream.from_app(
        self,
        asgi_routes=asgi_routes,
        asyncapi_path=asyncapi_path,
    )