Skip to content

PublisherUsecase

faststream.broker.publisher.usecase.PublisherUsecase #

PublisherUsecase(*, broker_middlewares, middlewares, schema_, title_, description_, include_in_schema)

Bases: ABC, AsyncAPIOperation, PublisherProto[MsgType]

A base class for publishers in an asynchronous API.

Source code in faststream/broker/publisher/usecase.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    *,
    broker_middlewares: Annotated[
        Sequence["BrokerMiddleware[MsgType]"],
        Doc("Top-level middlewares to use in direct `.publish` call."),
    ],
    middlewares: Annotated[
        Sequence["PublisherMiddleware"],
        Doc("Publisher middlewares."),
    ],
    # AsyncAPI args
    schema_: Annotated[
        Optional[Any],
        Doc(
            "AsyncAPI publishing message type"
            "Should be any python-native object annotation or `pydantic.BaseModel`."
        ),
    ],
    title_: Annotated[
        Optional[str],
        Doc("AsyncAPI object title."),
    ],
    description_: Annotated[
        Optional[str],
        Doc("AsyncAPI object description."),
    ],
    include_in_schema: Annotated[
        bool,
        Doc("Whetever to include operation in AsyncAPI schema or not."),
    ],
) -> None:
    self.calls = []
    self._middlewares = middlewares
    self._broker_middlewares = broker_middlewares
    self._producer = None

    self._fake_handler = False
    self.mock = None

    # AsyncAPI
    self.title_ = title_
    self.description_ = description_
    self.include_in_schema = include_in_schema
    self.schema_ = schema_

name property #

name

Returns the name of the API operation.

description property #

description

Returns the description of the API operation.

calls instance-attribute #

calls = []

mock instance-attribute #

mock = None

title_ instance-attribute #

title_ = title_

description_ instance-attribute #

description_ = description_

include_in_schema instance-attribute #

include_in_schema = include_in_schema

schema_ instance-attribute #

schema_ = schema_

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."""
    ...

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 {}

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

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)

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

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

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