Skip to content

create_publisher

faststream.redis.publisher.factory.create_publisher #

create_publisher(
    *,
    channel: Union[PubSub, str, None],
    list: Union[ListSub, str, None],
    stream: Union[StreamSub, str, None],
    headers: dict[str, Any] | None,
    reply_to: str,
    config: RedisBrokerConfig,
    message_format: type[MessageFormat] | None,
    title_: str | None,
    description_: str | None,
    schema_: Any | None,
    include_in_schema: bool,
) -> PublisherType
Source code in faststream/redis/publisher/factory.py 
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
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
def create_publisher(
    *,
    channel: Union["PubSub", str, None],
    list: Union["ListSub", str, None],
    stream: Union["StreamSub", str, None],
    headers: dict[str, Any] | None,
    reply_to: str,
    config: "RedisBrokerConfig",
    message_format: type["MessageFormat"] | None,
    # AsyncAPI args
    title_: str | None,
    description_: str | None,
    schema_: Any | None,
    include_in_schema: bool,
) -> PublisherType:
    validate_options(channel=channel, list=list, stream=stream)

    publisher_config = RedisPublisherConfig(
        reply_to=reply_to,
        headers=headers,
        _message_format=message_format,
        _outer_config=config,
    )

    specification_config = RedisPublisherSpecificationConfig(
        schema_=schema_,
        title_=title_,
        description_=description_,
        include_in_schema=include_in_schema,
    )

    specification: RedisPublisherSpecification
    if channel_sub := PubSub.validate(channel):
        specification = ChannelPublisherSpecification(
            config,
            specification_config,
            channel_sub,
        )

        return ChannelPublisher(publisher_config, specification, channel=channel_sub)

    if stream_sub := StreamSub.validate(stream):
        specification = StreamPublisherSpecification(
            config,
            specification_config,
            stream_sub,
        )

        return StreamPublisher(publisher_config, specification, stream=stream_sub)

    if list_sub := ListSub.validate(list):
        specification = ListPublisherSpecification(config, specification_config, list_sub)

        if list_sub.batch:
            return ListBatchPublisher(publisher_config, specification, list=list_sub)

        return ListPublisher(publisher_config, specification, list=list_sub)

    raise SetupError(INCORRECT_SETUP_MSG)