Skip to content

grpc_tools

frequenz.repo.config.setuptools.grpc_tools ¤

Setuptool hooks to build protobuf files.

This module contains a setuptools command that can be used to compile protocol buffer files in a project.

It also runs the command as the first sub-command for the build command, so protocol buffer files are compiled automatically before the project is built.

Classes¤

frequenz.repo.config.setuptools.grpc_tools.CompileProto ¤

Bases: Command

Build the Python protobuf files.

Source code in frequenz/repo/config/setuptools/grpc_tools.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
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class CompileProto(_setuptools.Command):
    """Build the Python protobuf files."""

    proto_path: str
    """The path of the root directory containing the protobuf files."""

    proto_glob: str
    """The glob pattern to use to find the protobuf files."""

    include_paths: str | Iterable[str]
    """Iterable or comma-separated list of paths to include when compiling the protobuf files."""

    py_path: str
    """The path of the root directory where the Python files will be generated."""

    description: str = "compile protobuf files"
    """Description of the command."""

    # We need the cast here because Command.user_options has the type annoatation
    # ClassVar[list[tuple[str, str, str]] | list[tuple[str, str | None, str]]] but the
    # expression resolves to list[tuple[str, None, str]] and mypy is not smart enough to
    # see that this is compatible with the list[tuple[str, str | None, str]] variant.
    user_options = cast(
        list[tuple[str, str, str]] | list[tuple[str, str | None, str]],
        [
            (
                "proto-path=",
                None,
                "path of the root directory containing the protobuf files",
            ),
            ("proto-glob=", None, "glob pattern to use to find the protobuf files"),
            (
                "include-paths=",
                None,
                "comma-separated list of paths to include when compiling the protobuf files",
            ),
            (
                "py-path=",
                None,
                "path of the root directory where the Python files will be generated",
            ),
        ],
    )
    """Options of the command."""

    def initialize_options(self) -> None:
        """Initialize options."""
        config = _protobuf.ProtobufConfig.from_pyproject_toml()

        self.proto_path = config.proto_path
        self.proto_glob = config.proto_glob
        self.include_paths = config.include_paths
        self.py_path = config.py_path

    def finalize_options(self) -> None:
        """Finalize options."""

    def run(self) -> None:
        """Compile the Python protobuf files."""
        include_paths: Iterable[str]
        match self.include_paths:
            case str() as str_paths:
                # If it comes as a comma-separated string, split it into a list,
                # stripping whitespace and ignoring empty strings.
                include_paths = filter(len, map(str.strip, str_paths.split(",")))
            case Iterable() as paths_it:
                include_paths = paths_it
            case unexpected:
                assert_never(unexpected)

        proto_files = [
            str(p) for p in _pathlib.Path(self.proto_path).rglob(self.proto_glob)
        ]

        if not proto_files:
            print(
                f"No proto files found in {self.proto_path}/**/{self.proto_glob}/, "
                "skipping compilation of proto files."
            )
            return

        protoc_cmd = (
            [_sys.executable, "-m", "grpc_tools.protoc"]
            + [f"-I{p}" for p in [*include_paths, self.proto_path]]
            + [
                f"--{opt}={self.py_path}"
                for opt in "python_out grpc_python_out mypy_out mypy_grpc_out".split()
            ]
            + proto_files
        )

        print(f"Compiling proto files via: {' '.join(protoc_cmd)}")
        _subprocess.run(protoc_cmd, check=True)
Attributes¤
description class-attribute instance-attribute ¤
description: str = 'compile protobuf files'

Description of the command.

include_paths instance-attribute ¤
include_paths: str | Iterable[str]

Iterable or comma-separated list of paths to include when compiling the protobuf files.

proto_glob instance-attribute ¤
proto_glob: str

The glob pattern to use to find the protobuf files.

proto_path instance-attribute ¤
proto_path: str

The path of the root directory containing the protobuf files.

py_path instance-attribute ¤
py_path: str

The path of the root directory where the Python files will be generated.

user_options class-attribute instance-attribute ¤
user_options = cast(
    list[tuple[str, str, str]]
    | list[tuple[str, str | None, str]],
    [
        (
            "proto-path=",
            None,
            "path of the root directory containing the protobuf files",
        ),
        (
            "proto-glob=",
            None,
            "glob pattern to use to find the protobuf files",
        ),
        (
            "include-paths=",
            None,
            "comma-separated list of paths to include when compiling the protobuf files",
        ),
        (
            "py-path=",
            None,
            "path of the root directory where the Python files will be generated",
        ),
    ],
)

Options of the command.

Functions¤
finalize_options ¤
finalize_options() -> None

Finalize options.

Source code in frequenz/repo/config/setuptools/grpc_tools.py 
79
80
def finalize_options(self) -> None:
    """Finalize options."""
initialize_options ¤
initialize_options() -> None

Initialize options.

Source code in frequenz/repo/config/setuptools/grpc_tools.py 
70
71
72
73
74
75
76
77
def initialize_options(self) -> None:
    """Initialize options."""
    config = _protobuf.ProtobufConfig.from_pyproject_toml()

    self.proto_path = config.proto_path
    self.proto_glob = config.proto_glob
    self.include_paths = config.include_paths
    self.py_path = config.py_path
run ¤
run() -> None

Compile the Python protobuf files.

Source code in frequenz/repo/config/setuptools/grpc_tools.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def run(self) -> None:
    """Compile the Python protobuf files."""
    include_paths: Iterable[str]
    match self.include_paths:
        case str() as str_paths:
            # If it comes as a comma-separated string, split it into a list,
            # stripping whitespace and ignoring empty strings.
            include_paths = filter(len, map(str.strip, str_paths.split(",")))
        case Iterable() as paths_it:
            include_paths = paths_it
        case unexpected:
            assert_never(unexpected)

    proto_files = [
        str(p) for p in _pathlib.Path(self.proto_path).rglob(self.proto_glob)
    ]

    if not proto_files:
        print(
            f"No proto files found in {self.proto_path}/**/{self.proto_glob}/, "
            "skipping compilation of proto files."
        )
        return

    protoc_cmd = (
        [_sys.executable, "-m", "grpc_tools.protoc"]
        + [f"-I{p}" for p in [*include_paths, self.proto_path]]
        + [
            f"--{opt}={self.py_path}"
            for opt in "python_out grpc_python_out mypy_out mypy_grpc_out".split()
        ]
        + proto_files
    )

    print(f"Compiling proto files via: {' '.join(protoc_cmd)}")
    _subprocess.run(protoc_cmd, check=True)