ObjStoreWatchSubscriber

faststream.nats.subscriber.ObjStoreWatchSubscriber

ObjStoreWatchSubscriber(
    config, specification, calls, *, obj_watch
)

Bases: TasksMixin, LogicSubscriber[ObjectInfo]

Source code in faststream/nats/subscriber/usecases/object_storage_subscriber.py

def __init__(
    self,
    config: "NatsSubscriberConfig",
    specification: "SubscriberSpecification[Any, Any]",
    calls: "CallsCollection[ObjectInfo]",
    *,
    obj_watch: "ObjWatch",
) -> None:
    parser = ObjParser(pattern="")
    config.parser = parser.parse_message
    config.decoder = parser.decode_message
    super().__init__(config, specification, calls)

    self.obj_watch = obj_watch
    self.obj_watch_conn = None

lock instance-attribute

lock = FakeContext()

extra_watcher_options instance-attribute

extra_watcher_options = {}

graceful_timeout instance-attribute

graceful_timeout

calls instance-attribute

calls = calls

specification instance-attribute

specification = specification

ack_policy instance-attribute

ack_policy = ack_policy

running instance-attribute

running = False

config instance-attribute

config = sub_config

extra_options instance-attribute

extra_options = extra_options or {}

subject property

subject

filter_subjects property

filter_subjects

clear_subject property

clear_subject

Compile test.{name} to test.* subject.

connection property

connection

jetstream property

jetstream

tasks instance-attribute

tasks = []

subscription instance-attribute

subscription

obj_watch instance-attribute

obj_watch = obj_watch

obj_watch_conn instance-attribute

obj_watch_conn = None

start async

start()

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()

stop async

stop()

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

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

consume async

consume(msg)

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

process_message async

process_message(msg)

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)

schema

schema()

Source code in faststream/_internal/endpoint/subscriber/usecase.py

def schema(self) -> dict[str, "SubscriberSpec"]:
    self._build_fastdepends_model()
    return self.specification.get_schema()

build_log_context 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_task

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

get_one async

get_one(*, timeout=5)

Source code in faststream/nats/subscriber/usecases/object_storage_subscriber.py

@override
async def get_one(self, *, timeout: float = 5) -> Optional["NatsObjMessage"]:
    assert not self.calls, (
        "You can't use `get_one` method if subscriber has registered handlers."
    )

    if not self._fetch_sub:
        self.bucket = await self._outer_config.os_declarer.create_object_store(
            bucket=self.subject,
            declare=self.obj_watch.declare,
        )

        obj_watch = await self.bucket.watch(
            ignore_deletes=self.obj_watch.ignore_deletes,
            include_history=self.obj_watch.include_history,
            meta_only=self.obj_watch.meta_only,
        )
        fetch_sub = self._fetch_sub = UnsubscribeAdapter["ObjectStore.ObjectWatcher"](
            obj_watch
        )
    else:
        fetch_sub = self._fetch_sub

    msg = None
    sleep_interval = timeout / 10
    with anyio.move_on_after(timeout):
        while (  # noqa: ASYNC110
            msg := await fetch_sub.obj.updates(timeout)  # type: ignore[no-untyped-call]
        ) is None:
            await anyio.sleep(sleep_interval)

    context = self._outer_config.fd_config.context
    async_parser, async_decoder = self._get_parser_and_decoder()

    return cast(
        "NatsObjMessage",
        await process_msg(
            msg=msg,
            middlewares=(m(msg, context=context) for m in self._broker_middlewares),
            parser=async_parser,
            decoder=async_decoder,
        ),
    )

get_log_context

get_log_context(message)

Log context factory using in self.consume scope.

PARAMETER	DESCRIPTION
message	Message which we are building context for TYPE: Optional[StreamMessage[ObjectInfo]]

Source code in faststream/nats/subscriber/usecases/object_storage_subscriber.py

def get_log_context(
    self,
    message: Optional["StreamMessage[ObjectInfo]"],
) -> dict[str, str]:
    """Log context factory using in `self.consume` scope.

    Args:
        message: Message which we are building context for
    """
    return self.build_log_context(
        message=message,
        subject=self.subject,
    )