get_app_broker_channels

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

    Args:
        app: An instance of FastStream or StreamRouter.

    Returns:
        A dictionary of channel names and their corresponding Channel objects.

    Raises:
        AssertionError: If the app does not have a broker.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    channels = {}
    assert app.broker  # nosec B101

    for h in app.broker.handlers.values():
        channels.update(h.schema())

    for p in app.broker._publishers.values():
        channels.update(p.schema())

    return channels