KafkaRouter

def __init__(
    self,
    bootstrap_servers: Annotated[
        Union[str, Iterable[str]],
        Doc(
            """
        A `host[:port]` string (or list of `host[:port]` strings) that the consumer should contact to bootstrap
        initial cluster metadata.

        This does not have to be the full node list.
        It just needs to have at least one broker that will respond to a
        Metadata API Request. Default port is 9092.
        """
        ),
    ] = "localhost",
    *,
    # both
    request_timeout_ms: Annotated[
        int,
        Doc("Client request timeout in milliseconds."),
    ] = 40 * 1000,
    retry_backoff_ms: Annotated[
        int,
        Doc("Milliseconds to backoff when retrying on errors."),
    ] = 100,
    metadata_max_age_ms: Annotated[
        int,
        Doc(
            """
        The period of time in milliseconds after
        which we force a refresh of metadata even if we haven't seen any
        partition leadership changes to proactively discover any new
        brokers or partitions.
        """
        ),
    ] = 5 * 60 * 1000,
    connections_max_idle_ms: Annotated[
        int,
        Doc(
            """
         Close idle connections after the number
        of milliseconds specified by this config. Specifying `None` will
        disable idle checks.
        """
        ),
    ] = 9 * 60 * 1000,
    client_id: Annotated[
        Optional[str],
        Doc(
            """
        A name for this client. This string is passed in
        each request to servers and can be used to identify specific
        server-side log entries that correspond to this client. Also
        submitted to :class:`~.consumer.group_coordinator.GroupCoordinator`
        for logging with respect to consumer group administration.
        """
        ),
    ] = SERVICE_NAME,
    allow_auto_create_topics: Annotated[
        bool,
        Doc(
            """
        Allow automatic topic creation on the broker when subscribing to or assigning non-existent topics.
        """
        ),
    ] = True,
    config: Annotated[
        Optional["ConfluentConfig"],
        Doc(
            """
            Extra configuration for the confluent-kafka-python
            producer/consumer. See `confluent_kafka.Config <https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#kafka-client-configuration>`_.
            """
        ),
    ] = None,
    # publisher args
    acks: Annotated[
        Literal[0, 1, -1, "all"],
        Doc(
            """
        One of ``0``, ``1``, ``all``. The number of acknowledgments
        the producer requires the leader to have received before considering a
        request complete. This controls the durability of records that are
        sent. The following settings are common:

        * ``0``: Producer will not wait for any acknowledgment from the server
          at all. The message will immediately be added to the socket
          buffer and considered sent. No guarantee can be made that the
          server has received the record in this case, and the retries
          configuration will not take effect (as the client won't
          generally know of any failures). The offset given back for each
          record will always be set to -1.
        * ``1``: The broker leader will write the record to its local log but
          will respond without awaiting full acknowledgement from all
          followers. In this case should the leader fail immediately
          after acknowledging the record but before the followers have
          replicated it then the record will be lost.
        * ``all``: The broker leader will wait for the full set of in-sync
          replicas to acknowledge the record. This guarantees that the
          record will not be lost as long as at least one in-sync replica
          remains alive. This is the strongest available guarantee.

        If unset, defaults to ``acks=1``. If `enable_idempotence` is
        :data:`True` defaults to ``acks=all``.
        """
        ),
    ] = EMPTY,
    compression_type: Annotated[
        Optional[Literal["gzip", "snappy", "lz4", "zstd"]],
        Doc(
            """
        The compression type for all data generated bythe producer.
        Compression is of full batches of data, so the efficacy of batching
        will also impact the compression ratio (more batching means better
        compression).
        """
        ),
    ] = None,
    partitioner: Annotated[
        Union[
            str,
            Callable[
                [bytes, List[Partition], List[Partition]],
                Partition,
            ],
        ],
        Doc(
            """
        Callable used to determine which partition
        each message is assigned to. Called (after key serialization):
        ``partitioner(key_bytes, all_partitions, available_partitions)``.
        The default partitioner implementation hashes each non-None key
        using the same murmur2 algorithm as the Java client so that
        messages with the same key are assigned to the same partition.
        When a key is :data:`None`, the message is delivered to a random partition
        (filtered to partitions with available leaders only, if possible).
        """
        ),
    ] = "consistent_random",
    max_request_size: Annotated[
        int,
        Doc(
            """
        The maximum size of a request. This is also
        effectively a cap on the maximum record size. Note that the server
        has its own cap on record size which may be different from this.
        This setting will limit the number of record batches the producer
        will send in a single request to avoid sending huge requests.
        """
        ),
    ] = 1024 * 1024,
    linger_ms: Annotated[
        int,
        Doc(
            """
        The producer groups together any records that arrive
        in between request transmissions into a single batched request.
        Normally this occurs only under load when records arrive faster
        than they can be sent out. However in some circumstances the client
        may want to reduce the number of requests even under moderate load.
        This setting accomplishes this by adding a small amount of
        artificial delay; that is, if first request is processed faster,
        than `linger_ms`, producer will wait ``linger_ms - process_time``.
        """
        ),
    ] = 0,
    enable_idempotence: Annotated[
        bool,
        Doc(
            """
        When set to `True`, the producer will
        ensure that exactly one copy of each message is written in the
        stream. If `False`, producer retries due to broker failures,
        etc., may write duplicates of the retried message in the stream.
        Note that enabling idempotence acks to set to ``all``. If it is not
        explicitly set by the user it will be chosen.
        """
        ),
    ] = False,
    transactional_id: Optional[str] = None,
    transaction_timeout_ms: int = 60 * 1000,
    # broker base args
    graceful_timeout: Annotated[
        Optional[float],
        Doc(
            "Graceful shutdown timeout. Broker waits for all running subscribers completion before shut down."
        ),
    ] = 15.0,
    decoder: Annotated[
        Optional["CustomCallable"],
        Doc("Custom decoder object."),
    ] = None,
    parser: Annotated[
        Optional["CustomCallable"],
        Doc("Custom parser object."),
    ] = None,
    middlewares: Annotated[
        Iterable[
            Union[
                "BrokerMiddleware[Message]",
                "BrokerMiddleware[Tuple[Message, ...]]",
            ]
        ],
        Doc("Middlewares to apply to all broker publishers/subscribers."),
    ] = (),
    # AsyncAPI args
    security: Annotated[
        Optional["BaseSecurity"],
        Doc(
            "Security options to connect broker and generate AsyncAPI server security information."
        ),
    ] = None,
    asyncapi_url: Annotated[
        Optional[str],
        Doc("AsyncAPI hardcoded server addresses. Use `servers` if not specified."),
    ] = None,
    protocol: Annotated[
        Optional[str],
        Doc("AsyncAPI server protocol."),
    ] = None,
    protocol_version: Annotated[
        Optional[str],
        Doc("AsyncAPI server protocol version."),
    ] = "auto",
    description: Annotated[
        Optional[str],
        Doc("AsyncAPI server description."),
    ] = None,
    asyncapi_tags: Annotated[
        Optional[Iterable[Union["asyncapi.Tag", "asyncapi.TagDict"]]],
        Doc("AsyncAPI server tags."),
    ] = None,
    # logging args
    logger: Annotated[
        Optional["LoggerProto"],
        Doc("User specified logger to pass into Context and log service messages."),
    ] = EMPTY,
    log_level: Annotated[
        int,
        Doc("Service messages log level."),
    ] = logging.INFO,
    log_fmt: Annotated[
        Optional[str],
        Doc("Default logger log format."),
    ] = None,
    # StreamRouter options
    setup_state: Annotated[
        bool,
        Doc(
            "Whether to add broker to app scope in lifespan. "
            "You should disable this option at old ASGI servers."
        ),
    ] = True,
    schema_url: Annotated[
        Optional[str],
        Doc(
            "AsyncAPI schema url. You should set this option to `None` to disable AsyncAPI routes at all."
        ),
    ] = "/asyncapi",
    # FastAPI args
    prefix: Annotated[
        str,
        Doc("An optional path prefix for the router."),
    ] = "",
    tags: Annotated[
        Optional[List[Union[str, "Enum"]]],
        Doc(
            """
            A list of tags to be applied to all the *path operations* in this
            router.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Optional[Sequence["params.Depends"]],
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to all the
            *path and stream operations* in this router.

            Read more about it in the
            [FastAPI docs for Bigger Applications - Multiple Files](https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies).
            """
        ),
    ] = None,
    default_response_class: Annotated[
        Type["Response"],
        Doc(
            """
            The default response class to be used.

            Read more in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#default-response-class).
            """
        ),
    ] = Default(JSONResponse),
    responses: Annotated[
        Optional[Dict[Union[int, str], "AnyDict"]],
        Doc(
            """
            Additional responses to be shown in OpenAPI.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Additional Responses in OpenAPI](https://fastapi.tiangolo.com/advanced/additional-responses/).

            And in the
            [FastAPI docs for Bigger Applications](https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies).
            """
        ),
    ] = None,
    callbacks: Annotated[
        Optional[List[BaseRoute]],
        Doc(
            """
            OpenAPI callbacks that should apply to all *path operations* in this
            router.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    routes: Annotated[
        Optional[List[BaseRoute]],
        Doc(
            """
            **Note**: you probably shouldn't use this parameter, it is inherited
            from Starlette and supported for compatibility.

            ---

            A list of routes to serve incoming HTTP and WebSocket requests.
            """
        ),
        deprecated(
            """
            You normally wouldn't use this parameter with FastAPI, it is inherited
            from Starlette and supported for compatibility.

            In FastAPI, you normally would use the *path operation methods*,
            like `router.get()`, `router.post()`, etc.
            """
        ),
    ] = None,
    redirect_slashes: Annotated[
        bool,
        Doc(
            """
            Whether to detect and redirect slashes in URLs when the client doesn't
            use the same format.
            """
        ),
    ] = True,
    default: Annotated[
        Optional["ASGIApp"],
        Doc(
            """
            Default function handler for this router. Used to handle
            404 Not Found errors.
            """
        ),
    ] = None,
    dependency_overrides_provider: Annotated[
        Optional[Any],
        Doc(
            """
            Only used internally by FastAPI to handle dependency overrides.

            You shouldn't need to use it. It normally points to the `FastAPI` app
            object.
            """
        ),
    ] = None,
    route_class: Annotated[
        Type["APIRoute"],
        Doc(
            """
            Custom route (*path operation*) class to be used by this router.

            Read more about it in the
            [FastAPI docs for Custom Request and APIRoute class](https://fastapi.tiangolo.com/how-to/custom-request-and-route/#custom-apiroute-class-in-a-router).
            """
        ),
    ] = APIRoute,
    on_startup: Annotated[
        Optional[Sequence[Callable[[], Any]]],
        Doc(
            """
            A list of startup event handler functions.

            You should instead use the `lifespan` handlers.

            Read more in the [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/).
            """
        ),
    ] = None,
    on_shutdown: Annotated[
        Optional[Sequence[Callable[[], Any]]],
        Doc(
            """
            A list of shutdown event handler functions.

            You should instead use the `lifespan` handlers.

            Read more in the
            [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/).
            """
        ),
    ] = None,
    lifespan: Annotated[
        Optional["Lifespan[Any]"],
        Doc(
            """
            A `Lifespan` context manager handler. This replaces `startup` and
            `shutdown` functions with a single context manager.

            Read more in the
            [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/).
            """
        ),
    ] = None,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Mark all *path operations* in this router as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            To include (or not) all the *path operations* in this router in the
            generated OpenAPI.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-from-openapi).
            """
        ),
    ] = True,
    generate_unique_id_function: Annotated[
        Callable[["APIRoute"], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> None:
    super().__init__(
        bootstrap_servers=bootstrap_servers,
        client_id=client_id,
        request_timeout_ms=request_timeout_ms,
        retry_backoff_ms=retry_backoff_ms,
        metadata_max_age_ms=metadata_max_age_ms,
        connections_max_idle_ms=connections_max_idle_ms,
        allow_auto_create_topics=allow_auto_create_topics,
        acks=acks,
        config=config,
        compression_type=compression_type,
        partitioner=partitioner,
        max_request_size=max_request_size,
        linger_ms=linger_ms,
        enable_idempotence=enable_idempotence,
        transactional_id=transactional_id,
        transaction_timeout_ms=transaction_timeout_ms,
        # broker args
        graceful_timeout=graceful_timeout,
        decoder=decoder,
        parser=parser,
        middlewares=middlewares,
        schema_url=schema_url,
        setup_state=setup_state,
        # logger options
        logger=logger,
        log_level=log_level,
        log_fmt=log_fmt,
        # AsyncAPI options
        security=security,
        protocol=protocol,
        description=description,
        protocol_version=protocol_version,
        asyncapi_tags=asyncapi_tags,
        asyncapi_url=asyncapi_url,
        # FastAPI kwargs
        prefix=prefix,
        tags=tags,
        dependencies=dependencies,
        default_response_class=default_response_class,
        responses=responses,
        callbacks=callbacks,
        routes=routes,
        redirect_slashes=redirect_slashes,
        default=default,
        dependency_overrides_provider=dependency_overrides_provider,
        route_class=route_class,
        on_startup=on_startup,
        on_shutdown=on_shutdown,
        deprecated=deprecated,
        include_in_schema=include_in_schema,
        lifespan=lifespan,
        generate_unique_id_function=generate_unique_id_function,
    )