StreamMessage
faststream.broker.fastapi.StreamMessage #
StreamMessage(
- body: Optional[AnyDict] = None,
- headers: Optional[AnyDict] = None,
- path: Optional[AnyDict] = None,
-)
- Bases: Request
A class to represent a stream message.
| METHOD | DESCRIPTION |
|---|---|
__init__ | initializes the StreamMessage object |
get_session | returns a callable function that handles the session of the message |
Initialize a class instance.
| PARAMETER | DESCRIPTION |
|---|---|
body | The body of the request as a dictionary. Default is None. TYPE: |
headers | The headers of the request as a dictionary. Default is None. TYPE: |
Source code in faststream/broker/fastapi/route.py
| Bases: A class to represent a stream message.
Initialize a class instance.
Source code in |
| |
url property #
Creates a session for handling requests.
| PARAMETER | DESCRIPTION |
|---|---|
dependant | The dependant object representing the session. TYPE: |
dependency_overrides_provider | Optional provider for dependency overrides. |
| RETURNS | DESCRIPTION |
|---|---|
Callable[[StreamMessage[Any]], Awaitable[SendableMessage]] | A callable that takes a native message and returns an awaitable sendable message. |
| RAISES | DESCRIPTION |
|---|---|
AssertionError | If the dependant call is not defined. |
Note
This function is used to create a session for handling requests. It takes a dependant object, which represents the session, and a dependency overrides provider, which allows for overriding dependencies. It returns a callable that takes a native message and returns an awaitable sendable message. The session is created based on the dependant object and the message passed to the callable. The session is then used to call the function obtained from the dependant object, and the result is returned.
Source code in faststream/broker/fastapi/route.py
| Creates a session for handling requests.
NoteThis function is used to create a session for handling requests. It takes a dependant object, which represents the session, and a dependency overrides provider, which allows for overriding dependencies. It returns a callable that takes a native message and returns an awaitable sendable message. The session is created based on the dependant object and the message passed to the callable. The session is then used to call the function obtained from the dependant object, and the result is returned. Source code in |
| |
is_disconnected async #
Source code in starlette/requests.py
294 295 diff --git a/0.3/api/faststream/broker/fastapi/StreamRoute/index.html b/0.3/api/faststream/broker/fastapi/StreamRoute/index.html index 405d4f3770..4a6cec4535 100644 --- a/0.3/api/faststream/broker/fastapi/StreamRoute/index.html +++ b/0.3/api/faststream/broker/fastapi/StreamRoute/index.html @@ -23,8 +23,7 @@ dependency_overrides_provider: Optional[Any] = None, **handle_kwargs: Any ) - Bases: A class representing a stream route. Initialize a class instance.
Source code in |
| Bases: A class representing a stream route. Initialize a class instance.
Source code in |
| |
handler instance-attribute #
handler: HandlerCallWrapper[
diff --git a/0.3/api/faststream/broker/fastapi/route/StreamMessage/index.html b/0.3/api/faststream/broker/fastapi/route/StreamMessage/index.html
index 129e7c5cc9..dcce1cdb78 100644
--- a/0.3/api/faststream/broker/fastapi/route/StreamMessage/index.html
+++ b/0.3/api/faststream/broker/fastapi/route/StreamMessage/index.html
@@ -10,12 +10,12 @@
body[data-md-color-scheme="slate"] .gslide-title { color: var(--md-default-fg-color);}
body[data-md-color-scheme="slate"] .gslide-desc { color: var(--md-default-fg-color);}
StreamMessage
faststream.broker.fastapi.route.StreamMessage #
StreamMessage(
- body: Optional[AnyDict] = None,
- headers: Optional[AnyDict] = None,
- path: Optional[AnyDict] = None,
-)
-
Bases: Request
A class to represent a stream message.
METHOD DESCRIPTION __init__ initializes the StreamMessage object
get_session returns a callable function that handles the session of the message
Initialize a class instance.
PARAMETER DESCRIPTION body The body of the request as a dictionary. Default is None.
TYPE: Optional[AnyDict] DEFAULT: None
headers The headers of the request as a dictionary. Default is None.
TYPE: Optional[AnyDict] DEFAULT: None
Source code in faststream/broker/fastapi/route.py
Bases: Request
A class to represent a stream message.
METHOD DESCRIPTION __init__ initializes the StreamMessage object
get_session returns a callable function that handles the session of the message
Initialize a class instance.
PARAMETER DESCRIPTION body The body of the request as a dictionary.
headers The headers of the request as a dictionary.
TYPE: AnyDict
Source code in faststream/broker/fastapi/route.py
def __init__(
- self,
- body: Optional[AnyDict] = None,
- headers: Optional[AnyDict] = None,
- path: Optional[AnyDict] = None,
-) -> None:
- """Initialize a class instance.
-
- Args:
- body: The body of the request as a dictionary. Default is None.
- headers: The headers of the request as a dictionary. Default is None.
-
- Attributes:
- scope: A dictionary to store the scope of the request.
- _cookies: A dictionary to store the cookies of the request.
- _headers: A dictionary to store the headers of the request.
- _body: A dictionary to store the body of the request.
- _query_params: A dictionary to store the query parameters of the request.
-
- """
- self.scope = {"path_params": path or {}}
- self._cookies = {}
- self._headers = headers or {}
- self._body = body or {}
- self._query_params = {**self._body, **(path or {})}
+171
+172
+173
+174
def __init__(
+ self,
+ *,
+ body: Union[AnyDict, List[Any]],
+ headers: AnyDict,
+ path: AnyDict,
+) -> None:
+ """Initialize a class instance.
+
+ Args:
+ body: The body of the request as a dictionary.
+ headers: The headers of the request as a dictionary.
+
+ Attributes:
+ scope: A dictionary to store the scope of the request.
+ _cookies: A dictionary to store the cookies of the request.
+ _headers: A dictionary to store the headers of the request.
+ _body: A dictionary to store the body of the request.
+ _query_params: A dictionary to store the query parameters of the request.
+
+ """
+ self._headers = headers
+ self._body = body
+ self._query_params = path
+
+ self.scope = {"path_params": self._query_params}
+ self._cookies = {}
url property #
Creates a session for handling requests.
PARAMETER DESCRIPTION dependant The dependant object representing the session.
TYPE: Dependant
dependency_overrides_provider Optional provider for dependency overrides.
RETURNS DESCRIPTION Callable[[StreamMessage[Any]], Awaitable[SendableMessage]] A callable that takes a native message and returns an awaitable sendable message.
RAISES DESCRIPTION AssertionError If the dependant call is not defined.
Note
This function is used to create a session for handling requests. It takes a dependant object, which represents the session, and a dependency overrides provider, which allows for overriding dependencies. It returns a callable that takes a native message and returns an awaitable sendable message. The session is created based on the dependant object and the message passed to the callable. The session is then used to call the function obtained from the dependant object, and the result is returned.
Source code in faststream/broker/fastapi/route.py
Creates a session for handling requests.
PARAMETER DESCRIPTION dependant The dependant object representing the session.
TYPE: Dependant
dependency_overrides_provider Optional provider for dependency overrides.
RETURNS DESCRIPTION Callable[[StreamMessage[Any]], Awaitable[SendableMessage]] A callable that takes a native message and returns an awaitable sendable message.
RAISES DESCRIPTION AssertionError If the dependant call is not defined.
Note
This function is used to create a session for handling requests. It takes a dependant object, which represents the session, and a dependency overrides provider, which allows for overriding dependencies. It returns a callable that takes a native message and returns an awaitable sendable message. The session is created based on the dependant object and the message passed to the callable. The session is then used to call the function obtained from the dependant object, and the result is returned.
Source code in faststream/broker/fastapi/route.py
@classmethod
-def get_session(
- cls,
- dependant: Dependant,
- dependency_overrides_provider: Optional[Any] = None,
-) -> Callable[[NativeMessage[Any]], Awaitable[SendableMessage]]:
- """Creates a session for handling requests.
-
- Args:
- dependant: The dependant object representing the session.
- dependency_overrides_provider: Optional provider for dependency overrides.
-
- Returns:
- A callable that takes a native message and returns an awaitable sendable message.
+236
+237
+238
+239
+240
+241
+242
+243
+244
+245
+246
+247
+248
+249
+250
+251
+252
+253
@classmethod
+def get_session(
+ cls,
+ dependant: Dependant,
+ dependency_overrides_provider: Optional[Any] = None,
+) -> Callable[[NativeMessage[Any]], Awaitable[SendableMessage]]:
+ """Creates a session for handling requests.
+
+ Args:
+ dependant: The dependant object representing the session.
+ dependency_overrides_provider: Optional provider for dependency overrides.
- Raises:
- AssertionError: If the dependant call is not defined.
+ Returns:
+ A callable that takes a native message and returns an awaitable sendable message.
- Note:
- This function is used to create a session for handling requests. It takes a dependant object, which represents the session, and a dependency overrides provider, which allows for overriding dependencies. It returns a callable that takes a native message and returns an awaitable sendable message. The session is created based on the dependant object and the message passed to the callable. The session is then used to call the function obtained from the dependant object, and the result is returned.
+ Raises:
+ AssertionError: If the dependant call is not defined.
- """
- assert dependant.call # nosec B101
+ Note:
+ This function is used to create a session for handling requests. It takes a dependant object, which represents the session, and a dependency overrides provider, which allows for overriding dependencies. It returns a callable that takes a native message and returns an awaitable sendable message. The session is created based on the dependant object and the message passed to the callable. The session is then used to call the function obtained from the dependant object, and the result is returned.
- func = get_app(dependant, dependency_overrides_provider)
-
- dependencies_names = tuple(i.name for i in dependant.dependencies)
-
- first_arg = next(
- dropwhile(
- lambda i: i in dependencies_names,
- inspect.signature(dependant.call).parameters,
- ),
- None,
- )
-
- async def app(message: NativeMessage[Any]) -> SendableMessage:
- """An asynchronous function that processes an incoming message and returns a sendable message.
+ """
+ assert dependant.call # nosec B101
+
+ func = get_app(dependant, dependency_overrides_provider)
+
+ dependencies_names = tuple(i.name for i in dependant.dependencies)
+
+ first_arg = next(
+ dropwhile(
+ lambda i: i in dependencies_names,
+ inspect.signature(dependant.call).parameters,
+ ),
+ None,
+ )
- Args:
- message : The incoming message to be processed
+ async def app(message: NativeMessage[Any]) -> SendableMessage:
+ """An asynchronous function that processes an incoming message and returns a sendable message.
- Returns:
- The sendable message
+ Args:
+ message : The incoming message to be processed
- Raises:
- TypeError: If the body of the message is not a dictionary
- !!! note
-
- The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
- """
- body = message.decoded_body
- if first_arg is not None:
- if not isinstance(body, dict) and not isinstance(body, list):
- fastapi_body: Any = {first_arg: body}
- else:
- fastapi_body = body
-
- session = cls(fastapi_body, message.headers, message.path)
- else:
- session = cls()
- return await func(session)
-
- return app
+ Returns:
+ The sendable message
+
+ Raises:
+ TypeError: If the body of the message is not a dictionary
+ !!! note
+
+ The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
+ """
+ body = message.decoded_body
+
+ fastapi_body: Union[AnyDict, List[Any]]
+ if first_arg is not None:
+ if isinstance(body, dict):
+ path = fastapi_body = body or {}
+ elif isinstance(body, list):
+ fastapi_body, path = body, {}
+ else:
+ path = fastapi_body = {first_arg: body}
+
+ session = cls(
+ body=fastapi_body,
+ headers=message.headers,
+ path={**path, **message.path},
+ )
+
+ else:
+ session = cls(
+ body={},
+ headers={},
+ path={},
+ )
+
+ return await func(session)
+
+ return app
is_disconnected async #
Source code in starlette/requests.py
294
295
diff --git a/0.3/api/faststream/broker/fastapi/route/StreamRoute/index.html b/0.3/api/faststream/broker/fastapi/route/StreamRoute/index.html
index 84e6c5f55e..0dc3ced94a 100644
--- a/0.3/api/faststream/broker/fastapi/route/StreamRoute/index.html
+++ b/0.3/api/faststream/broker/fastapi/route/StreamRoute/index.html
@@ -23,8 +23,7 @@
dependency_overrides_provider: Optional[Any] = None,
**handle_kwargs: Any
)
-
Bases: BaseRoute, Generic[MsgType, P_HandlerParams, T_HandlerReturn]
A class representing a stream route.
Initialize a class instance.
PARAMETER DESCRIPTION path The path of the instance.
TYPE: Union[NameRequired, str]
*extra Additional arguments.
TYPE: Union[NameRequired, str] DEFAULT: ()
endpoint The endpoint of the instance.
TYPE: Union[Callable[P_HandlerParams, T_HandlerReturn], HandlerCallWrapper[MsgType, P_HandlerParams, T_HandlerReturn]]
broker The broker of the instance.
TYPE: BrokerAsyncUsecase[MsgType, Any]
dependencies The dependencies of the instance.
TYPE: Sequence[Depends] DEFAULT: ()
dependency_overrides_provider The provider for dependency overrides.
**handle_kwargs Additional keyword arguments.
TYPE: Any DEFAULT: {}
RETURNS DESCRIPTION None None.
Source code in faststream/broker/fastapi/route.py
Bases: BaseRoute, Generic[MsgType, P_HandlerParams, T_HandlerReturn]
A class representing a stream route.
Initialize a class instance.
PARAMETER DESCRIPTION path The path of the instance.
TYPE: Union[NameRequired, str]
*extra Additional arguments.
TYPE: Union[NameRequired, str] DEFAULT: ()
endpoint The endpoint of the instance.
TYPE: Union[Callable[P_HandlerParams, T_HandlerReturn], HandlerCallWrapper[MsgType, P_HandlerParams, T_HandlerReturn]]
broker The broker of the instance.
TYPE: BrokerAsyncUsecase[MsgType, Any]
dependencies The dependencies of the instance.
TYPE: Sequence[Depends] DEFAULT: ()
dependency_overrides_provider The provider for dependency overrides.
**handle_kwargs Additional keyword arguments.
TYPE: Any DEFAULT: {}
RETURNS DESCRIPTION None None.
Source code in faststream/broker/fastapi/route.py
def __init__(
- self,
- path: Union[NameRequired, str],
- *extra: Union[NameRequired, str],
- endpoint: Union[
- Callable[P_HandlerParams, T_HandlerReturn],
- HandlerCallWrapper[MsgType, P_HandlerParams, T_HandlerReturn],
- ],
- broker: BrokerAsyncUsecase[MsgType, Any],
- dependencies: Sequence[params.Depends] = (),
- dependency_overrides_provider: Optional[Any] = None,
- **handle_kwargs: Any,
-) -> None:
- """Initialize a class instance.
-
- Args:
- path: The path of the instance.
- *extra: Additional arguments.
- endpoint: The endpoint of the instance.
- broker: The broker of the instance.
- dependencies: The dependencies of the instance.
- dependency_overrides_provider: The provider for dependency overrides.
- **handle_kwargs: Additional keyword arguments.
-
- Returns:
- None.
-
- """
- self.path = path
- self.broker = broker
-
- path_name = (path if isinstance(path, str) else path.name) or ""
-
- if isinstance(endpoint, HandlerCallWrapper):
- orig_call = endpoint._original_call
- else:
- orig_call = endpoint
-
- dependant = get_dependant(
- path=path_name,
- call=orig_call,
- )
- for depends in dependencies[::-1]:
- dependant.dependencies.insert(
- 0,
- get_parameterless_sub_dependant(depends=depends, path=path_name),
- )
- self.dependant = dependant
-
- call = wraps(orig_call)(
- StreamMessage.get_session(
- dependant,
- dependency_overrides_provider,
- )
- )
-
- if isinstance(endpoint, HandlerCallWrapper):
- endpoint._original_call = call
- handler = endpoint
-
- else:
- handler = call
-
- self.handler = broker.subscriber(
- path,
- *extra,
- _raw=True,
- _get_dependant=lambda call: dependant,
- **handle_kwargs,
- )(
- handler # type: ignore[arg-type]
- )
+122
+123
def __init__(
+ self,
+ path: Union[NameRequired, str],
+ *extra: Union[NameRequired, str],
+ endpoint: Union[
+ Callable[P_HandlerParams, T_HandlerReturn],
+ HandlerCallWrapper[MsgType, P_HandlerParams, T_HandlerReturn],
+ ],
+ broker: BrokerAsyncUsecase[MsgType, Any],
+ dependencies: Sequence[params.Depends] = (),
+ dependency_overrides_provider: Optional[Any] = None,
+ **handle_kwargs: Any,
+) -> None:
+ """Initialize a class instance.
+
+ Args:
+ path: The path of the instance.
+ *extra: Additional arguments.
+ endpoint: The endpoint of the instance.
+ broker: The broker of the instance.
+ dependencies: The dependencies of the instance.
+ dependency_overrides_provider: The provider for dependency overrides.
+ **handle_kwargs: Additional keyword arguments.
+
+ Returns:
+ None.
+
+ """
+ self.path = path
+ self.broker = broker
+
+ path_name = (path if isinstance(path, str) else path.name) or ""
+
+ if isinstance(endpoint, HandlerCallWrapper):
+ orig_call = endpoint._original_call
+ else:
+ orig_call = endpoint
+
+ dependant = get_dependant(
+ path=path_name,
+ call=orig_call,
+ )
+ for depends in dependencies[::-1]:
+ dependant.dependencies.insert(
+ 0,
+ get_parameterless_sub_dependant(depends=depends, path=path_name),
+ )
+ self.dependant = dependant
+
+ call = wraps(orig_call)(
+ StreamMessage.get_session(
+ dependant,
+ dependency_overrides_provider,
+ )
+ )
+
+ if isinstance(endpoint, HandlerCallWrapper):
+ endpoint._original_call = call
+ handler = endpoint
+
+ else:
+ handler = call
+
+ self.handler = broker.subscriber(
+ path,
+ *extra,
+ _raw=True,
+ _get_dependant=lambda call: dependant,
+ **handle_kwargs,
+ )(
+ handler # type: ignore[arg-type]
+ )
handler instance-attribute #
handler: HandlerCallWrapper[
diff --git a/0.3/api/faststream/broker/fastapi/route/get_app/index.html b/0.3/api/faststream/broker/fastapi/route/get_app/index.html
index 4f46411e80..6d249a1149 100644
--- a/0.3/api/faststream/broker/fastapi/route/get_app/index.html
+++ b/0.3/api/faststream/broker/fastapi/route/get_app/index.html
@@ -15,24 +15,7 @@
) -> Callable[
[StreamMessage], Coroutine[Any, Any, SendableMessage]
]
-
Creates a FastAPI application.
PARAMETER DESCRIPTION dependant The dependant object that defines the endpoint function and its dependencies.
TYPE: Dependant
dependency_overrides_provider Optional provider for dependency overrides.
RETURNS DESCRIPTION Callable[[StreamMessage], Coroutine[Any, Any, SendableMessage]] The FastAPI application as a callable that takes a StreamMessage object as input and returns a SendableMessage coroutine.
RAISES DESCRIPTION AssertionError If the code reaches an unreachable state.
Source code in faststream/broker/fastapi/route.py
Creates a FastAPI application.
PARAMETER DESCRIPTION dependant The dependant object that defines the endpoint function and its dependencies.
TYPE: Dependant
dependency_overrides_provider Optional provider for dependency overrides.
RETURNS DESCRIPTION Callable[[StreamMessage], Coroutine[Any, Any, SendableMessage]] The FastAPI application as a callable that takes a StreamMessage object as input and returns a SendableMessage coroutine.
RAISES DESCRIPTION AssertionError If the code reaches an unreachable state.
Source code in faststream/broker/fastapi/route.py
def get_app(
- dependant: Dependant,
- dependency_overrides_provider: Optional[Any] = None,
-) -> Callable[[StreamMessage], Coroutine[Any, Any, SendableMessage]]:
- """Creates a FastAPI application.
-
- Args:
- dependant: The dependant object that defines the endpoint function and its dependencies.
- dependency_overrides_provider: Optional provider for dependency overrides.
-
- Returns:
- The FastAPI application as a callable that takes a StreamMessage object as input and returns a SendableMessage coroutine.
-
- Raises:
- AssertionError: If the code reaches an unreachable state.
-
- """
-
- async def app(request: StreamMessage) -> SendableMessage:
- """Handle an HTTP request and return a response.
-
- Args:
- request: The incoming HTTP request.
-
- Returns:
- The response to be sent back to the client.
+295
+296
+297
+298
+299
+300
+301
+302
+303
+304
+305
+306
+307
+308
+309
+310
+311
+312
def get_app(
+ dependant: Dependant,
+ dependency_overrides_provider: Optional[Any] = None,
+) -> Callable[[StreamMessage], Coroutine[Any, Any, SendableMessage]]:
+ """Creates a FastAPI application.
+
+ Args:
+ dependant: The dependant object that defines the endpoint function and its dependencies.
+ dependency_overrides_provider: Optional provider for dependency overrides.
- Raises:
- AssertionError: If the code reaches an unreachable point.
+ Returns:
+ The FastAPI application as a callable that takes a StreamMessage object as input and returns a SendableMessage coroutine.
- """
- async with AsyncExitStack() as stack:
- request.scope["fastapi_astack"] = stack
-
- solved_result = await solve_dependencies(
- request=request,
- body=request._body,
- dependant=dependant,
- dependency_overrides_provider=dependency_overrides_provider,
- )
+ Raises:
+ AssertionError: If the code reaches an unreachable state.
+
+ """
+
+ async def app(request: StreamMessage) -> SendableMessage:
+ """Handle an HTTP request and return a response.
+
+ Args:
+ request: The incoming HTTP request.
- values, errors, _, _2, _3 = solved_result
- if errors:
- raise_fastapi_validation_error(errors, request._body)
-
- return cast(
- SendableMessage,
- await run_endpoint_function(
- dependant=dependant,
- values=values,
- is_coroutine=asyncio.iscoroutinefunction(dependant.call),
- ),
- )
-
- raise AssertionError("unreachable")
-
- return app
+ Returns:
+ The response to be sent back to the client.
+
+ Raises:
+ AssertionError: If the code reaches an unreachable point.
+
+ """
+ async with AsyncExitStack() as stack:
+ request.scope["fastapi_astack"] = stack
+
+ solved_result = await solve_dependencies(
+ request=request,
+ body=request._body,
+ dependant=dependant,
+ dependency_overrides_provider=dependency_overrides_provider,
+ )
+
+ values, errors, _, _2, _3 = solved_result
+ if errors:
+ raise_fastapi_validation_error(errors, request._body)
+
+ return cast(
+ SendableMessage,
+ await run_endpoint_function(
+ dependant=dependant,
+ values=values,
+ is_coroutine=asyncio.iscoroutinefunction(dependant.call),
+ ),
+ )
+
+ raise AssertionError("unreachable")
+
+ return app