Skip to content

channel

frequenz.client.base.channel ¤

Handling of gRPC channels.

Attributes¤

frequenz.client.base.channel.ChannelT module-attribute ¤

ChannelT = TypeVar(
    "ChannelT", bound=AsyncContextManager[Any]
)

A grpclib or grpcio channel type.

Functions¤

frequenz.client.base.channel.parse_grpc_uri ¤

parse_grpc_uri(
    uri: str,
    channel_type: type[ChannelT],
    /,
    *,
    default_port: int = 9090,
) -> ChannelT

Create a grpclib client channel from a URI.

The URI must have the following format:

grpc://hostname[:port][?ssl=false]

A few things to consider about URI components:

  • If any other components are present in the URI, a ValueError is raised.
  • If the port is omitted, the default_port is used.
  • If a query parameter is passed many times, the last value is used.
  • The only supported query parameter is ssl, which must be a boolean value and defaults to false.
  • Boolean query parameters can be specified with the following values (case-insensitive): true, 1, on, false, 0, off.
PARAMETER DESCRIPTION
uri

The gRPC URI specifying the connection parameters.

TYPE: str

channel_type

The type of channel to create.

TYPE: type[ChannelT]

default_port

The default port number to use if the URI does not specify one.

TYPE: int DEFAULT: 9090

RETURNS DESCRIPTION
ChannelT

A grpclib client channel object.
RAISES DESCRIPTION
ValueError

If the URI is invalid or contains unexpected components.
Source code in frequenz/client/base/channel.py 
25
26
27
28
29
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
def parse_grpc_uri(
    uri: str, channel_type: type[ChannelT], /, *, default_port: int = 9090
) -> ChannelT:
    """Create a grpclib client channel from a URI.

    The URI must have the following format:

    ```
    grpc://hostname[:port][?ssl=false]
    ```

    A few things to consider about URI components:

    - If any other components are present in the URI, a [`ValueError`][] is raised.
    - If the port is omitted, the `default_port` is used.
    - If a query parameter is passed many times, the last value is used.
    - The only supported query parameter is `ssl`, which must be a boolean value and
      defaults to `false`.
    - Boolean query parameters can be specified with the following values
      (case-insensitive): `true`, `1`, `on`, `false`, `0`, `off`.

    Args:
        uri: The gRPC URI specifying the connection parameters.
        channel_type: The type of channel to create.
        default_port: The default port number to use if the URI does not specify one.

    Returns:
        A grpclib client channel object.

    Raises:
        ValueError: If the URI is invalid or contains unexpected components.
    """
    parsed_uri = urlparse(uri)
    if parsed_uri.scheme != "grpc":
        raise ValueError(
            f"Invalid scheme '{parsed_uri.scheme}' in the URI, expected 'grpc'", uri
        )
    if not parsed_uri.hostname:
        raise ValueError(f"Host name is missing in URI '{uri}'", uri)
    for attr in ("path", "fragment", "params", "username", "password"):
        if getattr(parsed_uri, attr):
            raise ValueError(
                f"Unexpected {attr} '{getattr(parsed_uri, attr)}' in the URI '{uri}'",
                uri,
            )

    options = {k: v[-1] for k, v in parse_qs(parsed_uri.query).items()}
    ssl = _to_bool(options.pop("ssl", "false"))
    if options:
        raise ValueError(
            f"Unexpected query parameters {options!r} in the URI '{uri}'",
            uri,
        )

    host = parsed_uri.hostname
    port = parsed_uri.port or default_port
    match channel_type:
        case _grpchacks.GrpcioChannel:
            return cast(ChannelT, _grpchacks.grpcio_create_channel(host, port, ssl))
        case _grpchacks.GrpclibChannel:
            return cast(ChannelT, _grpchacks.grpclib_create_channel(host, port, ssl))
    assert False, "Unexpected channel type: {channel_type}"