Bases: TasksMixin, SubscriberUsecase[UnifyRedisDict]
A class to represent a Redis handler.
Source code in faststream/redis/subscriber/usecases/basic.py
| def __init__(
self,
config: "RedisSubscriberConfig",
specification: "SubscriberSpecification[Any, Any]",
calls: "CallsCollection[Any]",
) -> None:
super().__init__(config, specification, calls)
self.config = config
|
extra_watcher_options = {}
instance-attribute
instance-attribute
specification = specification
instance-attribute
instance-attribute
instance-attribute
async
Clean up handler subscription, cancel consume task in graceful mode.
Source code in faststream/_internal/endpoint/subscriber/mixins.py
| async def stop(self) -> None:
"""Clean up handler subscription, cancel consume task in graceful mode."""
await super().stop()
for task in self.tasks:
if not task.done():
task.cancel()
self.tasks.clear()
|
add_call(*, parser_, decoder_, middlewares_, dependencies_)
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| def add_call(
self,
*,
parser_: Optional["CustomCallable"],
decoder_: Optional["CustomCallable"],
middlewares_: Sequence["SubscriberMiddleware[Any]"],
dependencies_: Iterable["Dependant"],
) -> Self:
self._call_options = _CallOptions(
parser=parser_,
decoder=decoder_,
middlewares=middlewares_,
dependencies=dependencies_,
)
return self
|
async
Consume a message asynchronously.
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| async def consume(self, msg: MsgType) -> Any:
"""Consume a message asynchronously."""
if not self.running:
return None
try:
return await self.process_message(msg)
except StopConsume:
# Stop handler at StopConsume exception
await self.stop()
except SystemExit:
# Stop handler at `exit()` call
await self.stop()
if app := self._outer_config.fd_config.context.get("app"):
app.exit()
except Exception: # nosec B110
# All other exceptions were logged by CriticalLogMiddleware
pass
|
async
Execute all message processing stages.
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| async def process_message(self, msg: MsgType) -> "Response":
"""Execute all message processing stages."""
context = self._outer_config.fd_config.context
logger_state = self._outer_config.logger
async with AsyncExitStack() as stack:
stack.enter_context(self.lock)
# Enter context before middlewares
stack.enter_context(context.scope("handler_", self))
stack.enter_context(context.scope("logger", logger_state.logger.logger))
for k, v in self._outer_config.extra_context.items():
stack.enter_context(context.scope(k, v))
# enter all middlewares
middlewares: list[BaseMiddleware] = []
for base_m in self.__build__middlewares_stack():
middleware = base_m(msg, context=context)
middlewares.append(middleware)
await middleware.__aenter__()
cache: dict[Any, Any] = {}
parsing_error: Exception | None = None
for h in self.calls:
try:
message = await h.is_suitable(msg, cache)
except Exception as e:
parsing_error = e
break
if message is not None:
stack.enter_context(
context.scope("log_context", self.get_log_context(message)),
)
stack.enter_context(context.scope("message", message))
# Middlewares should be exited before scope release
for m in middlewares:
stack.push_async_exit(m.__aexit__)
result_msg = ensure_response(
await h.call(
message=message,
# consumer middlewares
_extra_middlewares=(
m.consume_scope for m in middlewares[::-1]
),
),
)
if not result_msg.correlation_id:
result_msg.correlation_id = message.correlation_id
for p in chain(
self.__get_response_publisher(message),
h.handler._publishers,
):
await p._publish(
result_msg.as_publish_command(),
_extra_middlewares=(
m.publish_scope for m in middlewares[::-1]
),
)
# Return data for tests
return result_msg
# Suitable handler was not found or
# parsing/decoding exception occurred
for m in middlewares:
stack.push_async_exit(m.__aexit__)
# Reraise it to catch in tests
if parsing_error:
raise parsing_error
error_msg = f"There is no suitable handler for {msg=}"
raise SubscriberNotFound(error_msg)
# An error was raised and processed by some middleware
return ensure_response(None)
|
abstractmethod
async
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| @abstractmethod
async def get_one(self, *, timeout: float = 5) -> Optional["StreamMessage[MsgType]"]:
raise NotImplementedError
|
Generate log context.
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| def get_log_context(
self,
message: Optional["StreamMessage[MsgType]"],
) -> dict[str, str]:
"""Generate log context."""
return {
"message_id": getattr(message, "message_id", ""),
}
|
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| def schema(self) -> dict[str, "SubscriberSpec"]:
self._build_fastdepends_model()
return self.specification.get_schema()
|
add_task(func, func_args=None, func_kwargs=None)
Source code in faststream/_internal/endpoint/subscriber/mixins.py
| def add_task(
self,
func: Callable[..., Coroutine[Any, Any, Any]],
func_args: tuple[Any, ...] | None = None,
func_kwargs: dict[str, Any] | None = None,
) -> asyncio.Task[Any]:
args = func_args or ()
kwargs = func_kwargs or {}
task = asyncio.create_task(func(*args, **kwargs))
callback = TaskCallbackSupervisor(func, func_args, func_kwargs, self)
task.add_done_callback(callback)
self.tasks.append(task)
return task
|
async
Source code in faststream/redis/subscriber/usecases/basic.py
| @override
async def start(
self,
*args: Any,
) -> None:
if self.tasks:
return
await super().start()
self._post_start()
start_signal = anyio.Event()
if self.calls:
self.add_task(self._consume, args, {"start_signal": start_signal})
with anyio.fail_after(3.0):
await start_signal.wait()
else:
start_signal.set()
|
staticmethod
build_log_context(message, channel='')
Source code in faststream/redis/subscriber/usecases/basic.py
| @staticmethod
def build_log_context(
message: Optional["BrokerStreamMessage[Any]"],
channel: str = "",
) -> dict[str, str]:
return {
"channel": channel,
"message_id": getattr(message, "message_id", ""),
}
|
async
Source code in faststream/redis/subscriber/usecases/basic.py
| async def consume_one(self, msg: Any) -> None:
await self.consume(msg)
|