Bases: SubscriberUsecase[MsgType]
Basic class for all NATS Subscriber types (KeyValue, ObjectStorage, Core & JetStream).
Source code in faststream/nats/subscriber/usecases/basic.py
| def __init__(
self,
config: "NatsSubscriberConfig",
specification: "SubscriberSpecification[Any, Any]",
calls: "CallsCollection[MsgType]",
) -> None:
super().__init__(config, specification, calls)
self._subject = config.subject
self.config = config.sub_config
self.extra_options = config.extra_options or {}
self._fetch_sub = None
self.subscription = None
|
instance-attribute
extra_options = extra_options or {}
instance-attribute
property
Compile test.{name} to test.* subject.
extra_watcher_options = {}
instance-attribute
instance-attribute
specification = specification
instance-attribute
instance-attribute
async
Create NATS subscription and start consume tasks.
Source code in faststream/nats/subscriber/usecases/basic.py
| async def start(self) -> None:
"""Create NATS subscription and start consume tasks."""
await super().start()
if self.calls:
await self._create_subscription()
self._post_start()
|
async
Clean up handler subscription, cancel consume task in graceful mode.
Source code in faststream/nats/subscriber/usecases/basic.py
| async def stop(self) -> None:
"""Clean up handler subscription, cancel consume task in graceful mode."""
await super().stop()
if self.subscription is not None:
await self.subscription.unsubscribe()
self.subscription = None
if self._fetch_sub is not None:
await self._fetch_sub.unsubscribe()
self._fetch_sub = None
|
staticmethod
build_log_context(message, subject, *, queue='', stream='')
Static method to build log context out of self.consume scope.
Source code in faststream/nats/subscriber/usecases/basic.py
| @staticmethod
def build_log_context(
message: Optional["StreamMessage[MsgType]"],
subject: str,
*,
queue: str = "",
stream: str = "",
) -> dict[str, str]:
"""Static method to build log context out of `self.consume` scope."""
return {
"subject": subject,
"queue": queue,
"stream": stream,
"message_id": getattr(message, "message_id", ""),
}
|
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()
|