Skip to content

Visitor

faststream.redis.testing.Visitor #

Bases: Protocol

visit #

visit(*, channel, list, stream, sub)
Source code in faststream/redis/testing.py 
263
264
265
266
267
268
269
270
def visit(
    self,
    *,
    channel: Optional[str],
    list: Optional[str],
    stream: Optional[str],
    sub: "LogicSubscriber",
) -> Optional[str]: ...

get_message #

get_message(channel, body, sub)
Source code in faststream/redis/testing.py 
272
def get_message(self, channel: str, body: Any, sub: "LogicSubscriber") -> Any: ...