Skip to content

create_subscriber

faststream.redis.subscriber.factory.create_subscriber #

create_subscriber(*, channel, list, stream, no_ack=False, no_reply=False, retry=False, broker_dependencies=(), broker_middlewares=(), title_=None, description_=None, include_in_schema=True)
Source code in faststream/redis/subscriber/factory.py 
 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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def create_subscriber(
    *,
    channel: Union["PubSub", str, None],
    list: Union["ListSub", str, None],
    stream: Union["StreamSub", str, None],
    # Subscriber args
    no_ack: bool = False,
    no_reply: bool = False,
    retry: bool = False,
    broker_dependencies: Iterable["Depends"] = (),
    broker_middlewares: Sequence["BrokerMiddleware[UnifyRedisDict]"] = (),
    # AsyncAPI args
    title_: Optional[str] = None,
    description_: Optional[str] = None,
    include_in_schema: bool = True,
) -> SubsciberType:
    validate_options(channel=channel, list=list, stream=stream)

    if (channel_sub := PubSub.validate(channel)) is not None:
        return AsyncAPIChannelSubscriber(
            channel=channel_sub,
            # basic args
            no_ack=no_ack,
            no_reply=no_reply,
            retry=retry,
            broker_dependencies=broker_dependencies,
            broker_middlewares=broker_middlewares,
            # AsyncAPI args
            title_=title_,
            description_=description_,
            include_in_schema=include_in_schema,
        )

    elif (stream_sub := StreamSub.validate(stream)) is not None:
        if stream_sub.batch:
            return AsyncAPIStreamBatchSubscriber(
                stream=stream_sub,
                # basic args
                no_ack=no_ack,
                no_reply=no_reply,
                retry=retry,
                broker_dependencies=broker_dependencies,
                broker_middlewares=broker_middlewares,
                # AsyncAPI args
                title_=title_,
                description_=description_,
                include_in_schema=include_in_schema,
            )
        else:
            return AsyncAPIStreamSubscriber(
                stream=stream_sub,
                # basic args
                no_ack=no_ack,
                no_reply=no_reply,
                retry=retry,
                broker_dependencies=broker_dependencies,
                broker_middlewares=broker_middlewares,
                # AsyncAPI args
                title_=title_,
                description_=description_,
                include_in_schema=include_in_schema,
            )

    elif (list_sub := ListSub.validate(list)) is not None:
        if list_sub.batch:
            return AsyncAPIListBatchSubscriber(
                list=list_sub,
                # basic args
                no_ack=no_ack,
                no_reply=no_reply,
                retry=retry,
                broker_dependencies=broker_dependencies,
                broker_middlewares=broker_middlewares,
                # AsyncAPI args
                title_=title_,
                description_=description_,
                include_in_schema=include_in_schema,
            )
        else:
            return AsyncAPIListSubscriber(
                list=list_sub,
                # basic args
                no_ack=no_ack,
                no_reply=no_reply,
                retry=retry,
                broker_dependencies=broker_dependencies,
                broker_middlewares=broker_middlewares,
                # AsyncAPI args
                title_=title_,
                description_=description_,
                include_in_schema=include_in_schema,
            )

    else:
        raise SetupError(INCORRECT_SETUP_MSG)