Skip to content

DefaultPublisher

faststream.confluent.publisher.usecase.DefaultPublisher #

DefaultPublisher(*, key, topic, partition, headers, reply_to, broker_middlewares, middlewares, schema_, title_, description_, include_in_schema)

Bases: LogicPublisher[Message]

Source code in faststream/confluent/publisher/usecase.py
def __init__(
    self,
    *,
    key: Optional[bytes],
    topic: str,
    partition: Optional[int],
    headers: Optional[Dict[str, str]],
    reply_to: Optional[str],
    # Publisher args
    broker_middlewares: Iterable["BrokerMiddleware[Message]"],
    middlewares: Iterable["PublisherMiddleware"],
    # AsyncAPI args
    schema_: Optional[Any],
    title_: Optional[str],
    description_: Optional[str],
    include_in_schema: bool,
) -> None:
    super().__init__(
        topic=topic,
        partition=partition,
        headers=headers,
        reply_to=reply_to,
        # publisher args
        broker_middlewares=broker_middlewares,
        middlewares=middlewares,
        # AsyncAPI args
        schema_=schema_,
        title_=title_,
        description_=description_,
        include_in_schema=include_in_schema,
    )

    self.key = key

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_

mock instance-attribute #

mock = None

calls instance-attribute #

calls = []

topic instance-attribute #

topic = topic

partition instance-attribute #

partition = partition

reply_to instance-attribute #

reply_to = reply_to

headers instance-attribute #

headers = headers

key instance-attribute #

key = key

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/confluent/publisher/usecase.py
def add_prefix(self, prefix: str) -> None:
    self.topic = "".join((prefix, self.topic))

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 abstractmethod staticmethod #

create()

Abstract factory to create a real Publisher.

Source code in faststream/broker/publisher/proto.py
@staticmethod
@abstractmethod
def create() -> "PublisherProto[MsgType]":
    """Abstract factory to create a real Publisher."""
    ...

get_name abstractmethod #

get_name()

Name property fallback.

Source code in faststream/asyncapi/abc.py
@abstractmethod
def get_name(self) -> str:
    """Name property fallback."""
    raise NotImplementedError()

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 abstractmethod #

get_schema()

Generate AsyncAPI schema.

Source code in faststream/asyncapi/abc.py
@abstractmethod
def get_schema(self) -> Dict[str, Channel]:
    """Generate AsyncAPI schema."""
    raise NotImplementedError()

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

publish async #

publish(message, topic='', *, key=None, partition=None, timestamp_ms=None, headers=None, correlation_id=None, reply_to='', no_confirm=False, _extra_middlewares=())
Source code in faststream/confluent/publisher/usecase.py
@override
async def publish(
    self,
    message: "SendableMessage",
    topic: str = "",
    *,
    key: Optional[bytes] = None,
    partition: Optional[int] = None,
    timestamp_ms: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None,
    correlation_id: Optional[str] = None,
    reply_to: str = "",
    no_confirm: bool = False,
    # publisher specific
    _extra_middlewares: Iterable["PublisherMiddleware"] = (),
) -> Optional[Any]:
    assert self._producer, NOT_CONNECTED_YET  # nosec B101

    kwargs: AnyDict = {
        "key": key or self.key,
        # basic args
        "no_confirm": no_confirm,
        "topic": topic or self.topic,
        "partition": partition or self.partition,
        "timestamp_ms": timestamp_ms,
        "headers": headers or self.headers,
        "reply_to": reply_to or self.reply_to,
        "correlation_id": correlation_id or gen_cor_id(),
    }

    call: AsyncFunc = self._producer.publish

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

    return await call(message, **kwargs)

request async #

request(message, topic='', *, key=None, partition=None, timestamp_ms=None, headers=None, correlation_id=None, timeout=0.5, _extra_middlewares=())
Source code in faststream/confluent/publisher/usecase.py
@override
async def request(
    self,
    message: "SendableMessage",
    topic: str = "",
    *,
    key: Optional[bytes] = None,
    partition: Optional[int] = None,
    timestamp_ms: Optional[int] = None,
    headers: Optional[Dict[str, str]] = None,
    correlation_id: Optional[str] = None,
    timeout: float = 0.5,
    # publisher specific
    _extra_middlewares: Iterable["PublisherMiddleware"] = (),
) -> "KafkaMessage":
    return await super().request(
        message=message,
        topic=topic,
        key=key or self.key,
        partition=partition,
        timestamp_ms=timestamp_ms,
        headers=headers,
        correlation_id=correlation_id,
        timeout=timeout,
        _extra_middlewares=_extra_middlewares,
    )