Skip to content

FastStream

faststream.FastStream #

FastStream(
    *brokers: BrokerUsecase[Any, Any, Any],
    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 
45
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
def __init__(
    self,
    *brokers: "BrokerUsecase[Any, Any, Any]",
    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__(
        *brokers,
        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, Any]]

logger instance-attribute #

logger = logger

lifespan_context instance-attribute #

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

set_broker #

set_broker(broker: BrokerUsecase[Any, 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 
108
109
110
111
112
113
114
115
116
@deprecated(
    "This method is deprecated and will be removed in 0.8.0 Use `add_broker` instead."
)
def set_broker(self, broker: "BrokerUsecase[Any, Any, Any]") -> None:
    """Set already existed App object broker.

    Useful then you create/init broker in `on_startup` hook.
    """
    self.add_broker(broker)

add_broker #

add_broker(broker: BrokerUsecase[Any, Any, Any]) -> None
Source code in faststream/_internal/application.py 
118
119
120
121
122
123
124
125
def add_broker(self, broker: "BrokerUsecase[Any, Any, Any]") -> None:
    if broker in self.brokers:
        msg = f"Broker {broker} is already added"
        raise SetupError(msg)

    self.brokers.append(broker)
    self.schema.add_broker(broker)
    broker._update_fd_config(self.config)

start async #

start(**run_extra_options: SettingField) -> None

Executes startup hooks and start broker.

Source code in faststream/_internal/application.py 
219
220
221
222
223
224
225
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 
268
269
270
271
272
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 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
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 
321
322
323
324
325
326
327
328
329
330
331
332
333
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 
335
336
337
338
339
340
341
342
343
344
345
346
347
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 
349
350
351
352
353
354
355
356
357
358
359
360
361
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 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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 
100
101
102
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 
104
105
106
107
108
109
110
111
112
113
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,
    )