Skip to content

LogicPublisher

faststream.redis.publisher.usecase.LogicPublisher #

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

Bases: PublisherUsecase[UnifyRedisDict]

A class to represent a Redis publisher.

Source code in faststream/redis/publisher/usecase.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    *,
    reply_to: str,
    headers: Optional["AnyDict"],
    # Publisher args
    broker_middlewares: Sequence["BrokerMiddleware[UnifyRedisDict]"],
    middlewares: Sequence["PublisherMiddleware"],
    # AsyncAPI args
    schema_: Optional[Any],
    title_: Optional[str],
    description_: Optional[str],
    include_in_schema: bool,
) -> None:
    super().__init__(
        broker_middlewares=broker_middlewares,
        middlewares=middlewares,
        # AsyncAPI args
        schema_=schema_,
        title_=title_,
        description_=description_,
        include_in_schema=include_in_schema,
    )

    self.reply_to = reply_to
    self.headers = headers

    self._producer = None

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 = []

reply_to instance-attribute #

reply_to = reply_to

headers instance-attribute #

headers = headers

publish abstractmethod async #

publish(message, /, *, correlation_id=None, _extra_middlewares=())

Publishes a message asynchronously.

Source code in faststream/broker/publisher/proto.py 
57
58
59
60
61
62
63
64
65
66
67
@abstractmethod
async def publish(
    self,
    message: "SendableMessage",
    /,
    *,
    correlation_id: Optional[str] = None,
    _extra_middlewares: Sequence["PublisherMiddleware"] = (),
) -> Optional[Any]:
    """Publishes a message asynchronously."""
    ...

request abstractmethod async #

request(message, /, *, correlation_id=None, _extra_middlewares=())

Publishes a message synchronously.

Source code in faststream/broker/publisher/proto.py 
69
70
71
72
73
74
75
76
77
78
79
@abstractmethod
async def request(
    self,
    message: "SendableMessage",
    /,
    *,
    correlation_id: Optional[str] = None,
    _extra_middlewares: Sequence["PublisherMiddleware"] = (),
) -> Optional[Any]:
    """Publishes a message synchronously."""
    ...

setup #

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

add_prefix abstractmethod #

add_prefix(prefix)
Source code in faststream/broker/proto.py 
11
12
@abstractmethod
def add_prefix(self, prefix: str) -> None: ...

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 
30
31
32
33
34
35
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 
94
95
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 
 97
 98
 99
100
101
@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 
16
17
18
19
@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 
26
27
28
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 
37
38
39
40
@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 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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 
121
122
123
124
def reset_test(self) -> None:
    """Turn off publisher's testing mode."""
    self._fake_handler = False
    self.mock = None

subscriber_property abstractmethod #

subscriber_property(*, name_only)
Source code in faststream/redis/publisher/usecase.py 
58
59
60
@abstractmethod
def subscriber_property(self, *, name_only: bool) -> "AnyDict":
    raise NotImplementedError()