get_app_broker_server

def get_app_broker_server(
    app: Union[FastStream, "StreamRouter[Any]"]
) -> Dict[str, Server]:
    """Get the broker server for an application.

    Args:
        app: An instance of `FastStream` or `StreamRouter` representing the application.

    Returns:
        A dictionary containing the broker servers. The keys are the server names and the values are instances of `Server` class.

    Raises:
        AssertionError: If the `broker` attribute of the app is not present.

    Note:
        This function is currently incomplete and the following fields in the `broker_meta` dictionary are not populated: "security", "variables", "bindings".
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    servers = {}

    broker = app.broker
    assert broker  # nosec B101

    broker_meta: Dict[str, Any] = {
        "protocol": broker.protocol,
        "protocolVersion": broker.protocol_version,
        "description": broker.description,
        "tags": broker.tags,
        # TODO
        # "variables": "",
        # "bindings": "",
    }

    if broker.security is not None:
        broker_meta["security"] = broker.security.get_requirement()

    if isinstance(broker.url, str):
        servers["development"] = Server(
            url=broker.url,
            **broker_meta,
        )

    elif len(broker.url) == 1:
        servers["development"] = Server(
            url=broker.url[0],
            **broker_meta,
        )

    else:
        for i, url in enumerate(broker.url, 1):
            servers[f"Server{i}"] = Server(
                url=url,
                **broker_meta,
            )

    return servers