Skip to content

AsyncAPIListBatchPublisher

faststream.redis.publisher.asyncapi.AsyncAPIListBatchPublisher #

AsyncAPIListBatchPublisher(*, list, reply_to, headers, broker_middlewares, middlewares, schema_, title_, description_, include_in_schema)

Bases: ListBatchPublisher, _ListPublisherMixin

Source code in faststream/redis/publisher/usecase.py
def __init__(
    self,
    *,
    list: "ListSub",
    reply_to: str,
    headers: Optional["AnyDict"],
    # Regular publisher options
    broker_middlewares: Iterable["BrokerMiddleware[UnifyRedisDict]"],
    middlewares: Iterable["PublisherMiddleware"],
    # AsyncAPI options
    schema_: Optional[Any],
    title_: Optional[str],
    description_: Optional[str],
    include_in_schema: bool,
) -> None:
    super().__init__(
        reply_to=reply_to,
        headers=headers,
        broker_middlewares=broker_middlewares,
        middlewares=middlewares,
        schema_=schema_,
        title_=title_,
        description_=description_,
        include_in_schema=include_in_schema,
    )

    self.list = list

title_ instance-attribute #

title_ = title_

description_ instance-attribute #

description_ = description_

include_in_schema instance-attribute #

include_in_schema = include_in_schema

name property #

name

Returns the name of the API operation.

description property #

description

Returns the description of the API operation.

schema_ instance-attribute #

schema_ = schema_

channel_binding property #

channel_binding

mock instance-attribute #

mock = None

calls instance-attribute #

calls = []

reply_to instance-attribute #

reply_to = reply_to

headers instance-attribute #

headers = headers

list instance-attribute #

list = list

publish async #

publish(message=(), list=None, *, correlation_id=None, headers=None, _extra_middlewares=(), **kwargs)
Source code in faststream/redis/publisher/usecase.py
@override
async def publish(  # type: ignore[override]
    self,
    message: Annotated[
        Iterable["SendableMessage"],
        Doc("Message body to send."),
    ] = (),
    list: Annotated[
        Optional[str],
        Doc("Redis List object name to send message."),
    ] = None,
    *,
    correlation_id: Annotated[
        Optional[str],
        Doc("Has no real effect. Option to be compatible with original protocol."),
    ] = None,
    headers: Annotated[
        Optional["AnyDict"],
        Doc("Message headers to store metainformation."),
    ] = None,
    # publisher specific
    _extra_middlewares: Annotated[
        Iterable["PublisherMiddleware"],
        Doc("Extra middlewares to wrap publishing process."),
    ] = (),
    **kwargs: Any,  # option to suppress maxlen
) -> None:
    assert self._producer, NOT_CONNECTED_YET  # nosec B101

    list_sub = ListSub.validate(list or self.list)
    correlation_id = correlation_id or gen_cor_id()

    call: AsyncFunc = self._producer.publish_batch

    for m in chain(
        (
            _extra_middlewares
            or (m(None).publish_scope for m in self._broker_middlewares)
        ),
        self._middlewares,
    ):
        call = partial(m, call)

    await call(
        *message,
        list=list_sub.name,
        correlation_id=correlation_id,
        headers=headers or self.headers,
    )

request async #

request(message=None, list=None, *, correlation_id=None, headers=None, timeout=30.0, _extra_middlewares=())
Source code in faststream/redis/publisher/usecase.py
@override
async def request(
    self,
    message: Annotated[
        "SendableMessage",
        Doc("Message body to send."),
    ] = None,
    list: Annotated[
        Optional[str],
        Doc("Redis List object name to send message."),
    ] = None,
    *,
    correlation_id: Annotated[
        Optional[str],
        Doc(
            "Manual message **correlation_id** setter. "
            "**correlation_id** is a useful option to trace messages."
        ),
    ] = None,
    headers: Annotated[
        Optional["AnyDict"],
        Doc("Message headers to store metainformation."),
    ] = None,
    timeout: Annotated[
        Optional[float],
        Doc("RPC reply waiting time."),
    ] = 30.0,
    # publisher specific
    _extra_middlewares: Annotated[
        Iterable["PublisherMiddleware"],
        Doc("Extra middlewares to wrap publishing process."),
    ] = (),
) -> "RedisMessage":
    assert self._producer, NOT_CONNECTED_YET  # nosec B101

    kwargs = {
        "list": ListSub.validate(list or self.list).name,
        # basic args
        "headers": headers or self.headers,
        "correlation_id": correlation_id or gen_cor_id(),
        "timeout": timeout,
    }

    request: AsyncFunc = self._producer.request

    for pub_m in chain(
        (
            _extra_middlewares
            or (m(None).publish_scope for m in self._broker_middlewares)
        ),
        self._middlewares,
    ):
        request = partial(pub_m, request)

    published_msg = await request(
        message,
        **kwargs,
    )

    async with AsyncExitStack() as stack:
        return_msg: Callable[[RedisMessage], Awaitable[RedisMessage]] = return_input
        for m in self._broker_middlewares:
            mid = m(published_msg)
            await stack.enter_async_context(mid)
            return_msg = partial(mid.consume_scope, return_msg)

        parsed_msg = await self._producer._parser(published_msg)
        parsed_msg._decoded_body = await self._producer._decoder(parsed_msg)
        parsed_msg._source_type = SourceType.Response
        return await return_msg(parsed_msg)

    raise AssertionError("unreachable")

setup #

setup(*, producer)
Source code in faststream/broker/publisher/usecase.py
@override
def setup(  # type: ignore[override]
    self,
    *,
    producer: Optional["ProducerProto"],
) -> None:
    self._producer = producer

add_prefix #

add_prefix(prefix)
Source code in faststream/redis/publisher/usecase.py
def add_prefix(self, prefix: str) -> None:
    list_sub = deepcopy(self.list)
    list_sub.name = "".join((prefix, list_sub.name))
    self.list = list_sub

schema #

schema()

Returns the schema of the API operation as a dictionary of channel names and channel objects.

Source code in faststream/asyncapi/abc.py
def schema(self) -> Dict[str, Channel]:
    """Returns the schema of the API operation as a dictionary of channel names and channel objects."""
    if self.include_in_schema:
        return self.get_schema()
    else:
        return {}

add_middleware #

add_middleware(middleware)
Source code in faststream/broker/publisher/usecase.py
def add_middleware(self, middleware: "BrokerMiddleware[MsgType]") -> None:
    self._broker_middlewares = (*self._broker_middlewares, middleware)

create staticmethod #

create(*, channel, list, stream, headers, reply_to, broker_middlewares, middlewares, title_, description_, schema_, include_in_schema)
Source code in faststream/redis/publisher/asyncapi.py
@override
@staticmethod
def create(  # type: ignore[override]
    *,
    channel: Union["PubSub", str, None],
    list: Union["ListSub", str, None],
    stream: Union["StreamSub", str, None],
    headers: Optional["AnyDict"],
    reply_to: str,
    broker_middlewares: Iterable["BrokerMiddleware[UnifyRedisDict]"],
    middlewares: Iterable["PublisherMiddleware"],
    # AsyncAPI args
    title_: Optional[str],
    description_: Optional[str],
    schema_: Optional[Any],
    include_in_schema: bool,
) -> PublisherType:
    validate_options(channel=channel, list=list, stream=stream)

    if (channel := PubSub.validate(channel)) is not None:
        return AsyncAPIChannelPublisher(
            channel=channel,
            # basic args
            headers=headers,
            reply_to=reply_to,
            broker_middlewares=broker_middlewares,
            middlewares=middlewares,
            # AsyncAPI args
            title_=title_,
            description_=description_,
            schema_=schema_,
            include_in_schema=include_in_schema,
        )

    elif (stream := StreamSub.validate(stream)) is not None:
        return AsyncAPIStreamPublisher(
            stream=stream,
            # basic args
            headers=headers,
            reply_to=reply_to,
            broker_middlewares=broker_middlewares,
            middlewares=middlewares,
            # AsyncAPI args
            title_=title_,
            description_=description_,
            schema_=schema_,
            include_in_schema=include_in_schema,
        )

    elif (list := ListSub.validate(list)) is not None:
        if list.batch:
            return AsyncAPIListBatchPublisher(
                list=list,
                # basic args
                headers=headers,
                reply_to=reply_to,
                broker_middlewares=broker_middlewares,
                middlewares=middlewares,
                # AsyncAPI args
                title_=title_,
                description_=description_,
                schema_=schema_,
                include_in_schema=include_in_schema,
            )
        else:
            return AsyncAPIListPublisher(
                list=list,
                # basic args
                headers=headers,
                reply_to=reply_to,
                broker_middlewares=broker_middlewares,
                middlewares=middlewares,
                # AsyncAPI args
                title_=title_,
                description_=description_,
                schema_=schema_,
                include_in_schema=include_in_schema,
            )

    else:
        raise SetupError(INCORRECT_SETUP_MSG)

get_name #

get_name()
Source code in faststream/redis/publisher/asyncapi.py
def get_name(self) -> str:
    return f"{self.list.name}:Publisher"

get_description #

get_description()

Description property fallback.

Source code in faststream/asyncapi/abc.py
def get_description(self) -> Optional[str]:
    """Description property fallback."""
    return None

get_schema #

get_schema()
Source code in faststream/redis/publisher/asyncapi.py
def get_schema(self) -> Dict[str, Channel]:
    payloads = self.get_payloads()

    return {
        self.name: Channel(
            description=self.description,
            publish=Operation(
                message=Message(
                    title=f"{self.name}:Message",
                    payload=resolve_payloads(payloads, "Publisher"),
                    correlationId=CorrelationId(
                        location="$message.header#/correlation_id"
                    ),
                ),
            ),
            bindings=ChannelBinding(
                redis=self.channel_binding,
            ),
        )
    }

get_payloads #

get_payloads()
Source code in faststream/broker/publisher/usecase.py
def get_payloads(self) -> List[Tuple["AnyDict", str]]:
    payloads: List[Tuple[AnyDict, str]] = []

    if self.schema_:
        params = {"response__": (self.schema_, ...)}

        call_model: CallModel[Any, Any] = CallModel(
            call=lambda: None,
            model=create_model("Fake"),
            response_model=create_model(  # type: ignore[call-overload]
                "",
                __config__=get_config_base(),
                **params,
            ),
            params=params,
        )

        body = get_response_schema(
            call_model,
            prefix=f"{self.name}:Message",
        )
        if body:  # pragma: no branch
            payloads.append((body, ""))

    else:
        for call in self.calls:
            call_model = build_call_model(call)
            body = get_response_schema(
                call_model,
                prefix=f"{self.name}:Message",
            )
            if body:
                payloads.append((body, to_camelcase(unwrap(call).__name__)))

    return payloads

set_test #

set_test(*, mock, with_fake)

Turn publisher to testing mode.

Source code in faststream/broker/publisher/usecase.py
def set_test(
    self,
    *,
    mock: Annotated[
        MagicMock,
        Doc("Mock object to check in tests."),
    ],
    with_fake: Annotated[
        bool,
        Doc("Whetevet publisher's fake subscriber created or not."),
    ],
) -> None:
    """Turn publisher to testing mode."""
    self.mock = mock
    self._fake_handler = with_fake

reset_test #

reset_test()

Turn off publisher's testing mode.

Source code in faststream/broker/publisher/usecase.py
def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self._fake_handler = False
    self.mock = None

subscriber_property #

subscriber_property(*, name_only)
Source code in faststream/redis/publisher/usecase.py
@override
def subscriber_property(self, *, name_only: bool) -> "AnyDict":
    return {
        "channel": None,
        "list": self.list.name if name_only else self.list,
        "stream": None,
    }