channel

frequenz.client.base.channel

Handling of gRPC channels.

Classes

frequenz.client.base.channel.ChannelOptions dataclass

Options for a gRPC channel.

Source code in frequenz/client/base/channel.py

@dataclasses.dataclass(frozen=True)
class ChannelOptions:
    """Options for a gRPC channel."""

    port: int | None = None
    """The port number to connect to."""

    ssl: SslOptions = SslOptions()
    """SSL options for the channel."""

    keep_alive: KeepAliveOptions = KeepAliveOptions()
    """HTTP2 keep-alive options for the channel."""

Attributes

keep_alive class-attribute instance-attribute

keep_alive: KeepAliveOptions = KeepAliveOptions()

HTTP2 keep-alive options for the channel.

port class-attribute instance-attribute

port: int | None = None

The port number to connect to.

ssl class-attribute instance-attribute

ssl: SslOptions = SslOptions()

SSL options for the channel.

frequenz.client.base.channel.KeepAliveOptions dataclass

Options for HTTP2 keep-alive pings.

Source code in frequenz/client/base/channel.py

@dataclasses.dataclass(frozen=True)
class KeepAliveOptions:
    """Options for HTTP2 keep-alive pings."""

    enabled: bool = True
    """Whether HTTP2 keep-alive should be enabled."""

    interval: timedelta = timedelta(seconds=60)
    """The interval between HTTP2 pings."""

    timeout: timedelta = timedelta(seconds=20)
    """The time to wait for a HTTP2 keep-alive response."""

Attributes

enabled class-attribute instance-attribute

enabled: bool = True

Whether HTTP2 keep-alive should be enabled.

interval class-attribute instance-attribute

interval: timedelta = timedelta(seconds=60)

The interval between HTTP2 pings.

timeout class-attribute instance-attribute

timeout: timedelta = timedelta(seconds=20)

The time to wait for a HTTP2 keep-alive response.

frequenz.client.base.channel.SslOptions dataclass

SSL options for a gRPC channel.

Source code in frequenz/client/base/channel.py

@dataclasses.dataclass(frozen=True)
class SslOptions:
    """SSL options for a gRPC channel."""

    enabled: bool = True
    """Whether SSL should be enabled."""

    root_certificates: pathlib.Path | bytes | None = None
    """The PEM-encoded root certificates.

    This can be a path to a file containing the certificates, a byte string, or None to
    retrieve them from a default location chosen by gRPC runtime.
    """

    private_key: pathlib.Path | bytes | None = None
    """The PEM-encoded private key.

    This can be a path to a file containing the key, a byte string, or None if no key
    should be used.
    """

    certificate_chain: pathlib.Path | bytes | None = None
    """The PEM-encoded certificate chain.

    This can be a path to a file containing the chain, a byte string, or None if no
    chain should be used.
    """

Attributes

certificate_chain class-attribute instance-attribute

certificate_chain: Path | bytes | None = None

The PEM-encoded certificate chain.

This can be a path to a file containing the chain, a byte string, or None if no chain should be used.

enabled class-attribute instance-attribute

enabled: bool = True

Whether SSL should be enabled.

private_key class-attribute instance-attribute

private_key: Path | bytes | None = None

The PEM-encoded private key.

This can be a path to a file containing the key, a byte string, or None if no key should be used.

root_certificates class-attribute instance-attribute

root_certificates: Path | bytes | None = None

The PEM-encoded root certificates.

This can be a path to a file containing the certificates, a byte string, or None to retrieve them from a default location chosen by gRPC runtime.

Functions

frequenz.client.base.channel.parse_grpc_uri

parse_grpc_uri(
    uri: str, /, defaults: ChannelOptions = ChannelOptions()
) -> Channel

Create a client channel from a URI.

The URI must have the following format:

grpc://hostname[:port][?param=value&...]

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 unless it is None, in which case a ValueError is raised
• If a query parameter is passed many times, the last value is used.
• Boolean query parameters can be specified with the following values (case-insensitive): true, 1, on, false, 0, off.

Supported query parameters:

• ssl (bool): Enable or disable SSL. Defaults to default_ssl.
• ssl_root_certificates_path (str): Path to the root certificates file. Only valid if SSL is enabled. Will raise a ValueError if the file cannot be read.
• ssl_private_key_path (str): Path to the private key file. Only valid if SSL is enabled. Will raise a ValueError if the file cannot be read.
• ssl_certificate_chain_path (str): Path to the certificate chain file. Only valid if SSL is enabled. Will raise a ValueError if the file cannot be read.

PARAMETER	DESCRIPTION
uri	The gRPC URI specifying the connection parameters. TYPE: str
defaults	The default options use to create the channel when not specified in the URI. TYPE: ChannelOptions DEFAULT: ChannelOptions()

RETURNS	DESCRIPTION
Channel	A client channel object.

RAISES	DESCRIPTION
ValueError	If the URI is invalid or contains unexpected components.

Source code in frequenz/client/base/channel.py

def parse_grpc_uri(
    uri: str,
    /,
    defaults: ChannelOptions = ChannelOptions(),
) -> Channel:
    """Create a client channel from a URI.

    The URI must have the following format:

    ```
    grpc://hostname[:port][?param=value&...]
    ```

    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 unless it is `None`, in which
      case a `ValueError` is raised
    - If a query parameter is passed many times, the last value is used.
    - Boolean query parameters can be specified with the following values
      (case-insensitive): `true`, `1`, `on`, `false`, `0`, `off`.

    Supported query parameters:

    - `ssl` (bool): Enable or disable SSL. Defaults to `default_ssl`.
    - `ssl_root_certificates_path` (str): Path to the root certificates file. Only
      valid if SSL is enabled. Will raise a `ValueError` if the file cannot be read.
    - `ssl_private_key_path` (str): Path to the private key file. Only valid if SSL is
      enabled. Will raise a `ValueError` if the file cannot be read.
    - `ssl_certificate_chain_path` (str): Path to the certificate chain file. Only
      valid if SSL is enabled. Will raise a `ValueError` if the file cannot be read.

    Args:
        uri: The gRPC URI specifying the connection parameters.
        defaults: The default options use to create the channel when not specified in
            the URI.

    Returns:
        A 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 = _parse_query_params(uri, parsed_uri.query)

    if parsed_uri.port is None and defaults.port is None:
        raise ValueError(
            f"The gRPC URI '{uri}' doesn't specify a port and there is no default."
        )

    target = (
        parsed_uri.netloc if parsed_uri.port else f"{parsed_uri.netloc}:{defaults.port}"
    )

    keep_alive = (
        defaults.keep_alive.enabled
        if options.keep_alive is None
        else options.keep_alive
    )
    channel_options = (
        [
            ("grpc.http2.max_pings_without_data", 0),
            ("grpc.keepalive_permit_without_calls", 1),
            (
                "grpc.keepalive_time_ms",
                (
                    (
                        defaults.keep_alive.interval
                        if options.keep_alive_interval is None
                        else options.keep_alive_interval
                    ).total_seconds()
                    * 1000
                ),
            ),
            (
                "grpc.keepalive_timeout_ms",
                (
                    defaults.keep_alive.timeout
                    if options.keep_alive_timeout is None
                    else options.keep_alive_timeout
                ).total_seconds()
                * 1000,
            ),
        ]
        if keep_alive
        else None
    )

    ssl = defaults.ssl.enabled if options.ssl is None else options.ssl
    if ssl:
        return secure_channel(
            target,
            ssl_channel_credentials(
                root_certificates=_get_contents(
                    "root certificates",
                    options.ssl_root_certificates_path,
                    defaults.ssl.root_certificates,
                ),
                private_key=_get_contents(
                    "private key",
                    options.ssl_private_key_path,
                    defaults.ssl.private_key,
                ),
                certificate_chain=_get_contents(
                    "certificate chain",
                    options.ssl_certificate_chain_path,
                    defaults.ssl.certificate_chain,
                ),
            ),
            channel_options,
        )
    return insecure_channel(target, channel_options)