Bases: TasksMixin, SubscriberUsecase[MsgType]
A class to handle logic for consuming messages from Kafka.
Source code in faststream/confluent/subscriber/usecase.py
| def __init__(
self,
config: "KafkaSubscriberConfig",
specification: "SubscriberSpecification[Any, Any]",
calls: "CallsCollection[MsgType]",
) -> None:
super().__init__(config, specification, calls)
self.__connection_data = config.connection_data
self.group_id = config.group_id
self._topics = config.topics
self._partitions = config.partitions
self.consumer = None
self.polling_interval = config.polling_interval
|
instance-attribute
instance-attribute
instance-attribute
instance-attribute
polling_interval = polling_interval
extra_watcher_options = {}
instance-attribute
instance-attribute
specification = specification
instance-attribute
instance-attribute
async
Start the consumer.
Source code in faststream/confluent/subscriber/usecase.py
| @override
async def start(self) -> None:
"""Start the consumer."""
await super().start()
self.consumer = consumer = self._outer_config.builder(
*self.topics,
partitions=self.partitions,
group_id=self.group_id,
client_id=self.client_id,
**self.__connection_data,
)
self.parser._setup(consumer)
await consumer.start()
self._post_start()
if self.calls:
self.add_task(self._consume)
|
async
Source code in faststream/confluent/subscriber/usecase.py
| async def stop(self) -> None:
await super().stop()
if self.consumer is not None:
await self.consumer.stop()
self.consumer = None
|
async
Source code in faststream/confluent/subscriber/usecase.py
| @override
async def get_one(
self,
*,
timeout: float = 5.0,
) -> "KafkaMessage | None":
assert self.consumer, "You should start subscriber at first."
assert not self.calls, (
"You can't use `get_one` method if subscriber has registered handlers."
)
raw_message = await self.consumer.getone(timeout=timeout)
context = self._outer_config.fd_config.context
async_parser, async_decoder = self._get_parser_and_decoder()
return await process_msg( # type: ignore[return-value]
msg=raw_message,
middlewares=(
m(raw_message, context=context) for m in self._broker_middlewares
),
parser=async_parser,
decoder=async_decoder,
)
|
async
Source code in faststream/confluent/subscriber/usecase.py
| async def consume_one(self, msg: MsgType) -> None:
await self.consume(msg)
|
abstractmethod
async
Source code in faststream/confluent/subscriber/usecase.py
| @abstractmethod
async def get_msg(self) -> MsgType | None:
raise NotImplementedError
|
staticmethod
build_log_context(message, topic, group_id=None)
Source code in faststream/confluent/subscriber/usecase.py
| @staticmethod
def build_log_context(
message: Optional["StreamMessage[Any]"],
topic: str,
group_id: str | None = None,
) -> dict[str, str]:
return {
"topic": topic,
"group_id": group_id or "",
"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)
|
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
|