misc.python.materialize.build_config

  1# Copyright Materialize, Inc. and contributors. All rights reserved.
  2#
  3# Use of this software is governed by the Business Source License
  4# included in the LICENSE file at the root of this repository.
  5#
  6# As of the Change Date specified in that file, in accordance with
  7# the Business Source License, use of this software will be governed
  8# by the Apache License, Version 2.0.
  9
 10import os
 11from pathlib import Path
 12from textwrap import dedent
 13from typing import Any
 14
 15import toml
 16
 17
 18class BuildConfig:
 19    """Configuration for builds of Materialize.
 20
 21    Most things should be configured via a tool's native configuration file,
 22    e.g. `.bazelrc` or `.cargo/config.toml`. This exists for Materialize's home
 23    grown tools, or to extend tools that don't support an option we need.
 24
 25    Looks for configuration files in `~/.config/materialize/build.toml`
 26    """
 27
 28    def __init__(self, path: Path):
 29        if path.is_file():
 30            with open(path) as f:
 31                raw_data = toml.load(f)
 32        else:
 33            raw_data = {}
 34
 35        self.bazel = BazelConfig(raw_data.get("bazel", {}))
 36
 37    @staticmethod
 38    def read():
 39        home = Path.home()
 40        path = home / ".config" / "materialize" / "build.toml"
 41
 42        return BuildConfig(path)
 43
 44    def __str__(self):
 45        return f"{self.bazel}"
 46
 47
 48class BazelConfig:
 49    """Configuration for Bazel builds.
 50
 51    Most configuration should go into either the repositories `.bazelrc` file
 52    or documented to be included in a users' home `.bazelrc` file. This exists
 53    for flags that Bazel does not have an easy way to configure itself.
 54
 55    [bazel]
 56    remote_cache = "localhost:6889"
 57    """
 58
 59    def __init__(self, data: dict[str, Any]):
 60        self.remote_cache = data.get("remote_cache", None)
 61        self.remote_cache_check_interval_minutes = data.get(
 62            "remote_cache_check_interval_minutes", "5"
 63        )
 64
 65    def __str__(self):
 66        return dedent(
 67            f"""
 68        Bazel:
 69            remote_cache = {self.remote_cache}
 70        """
 71        )
 72
 73
 74class LocalState:
 75    """Local state persisted by a tool.
 76
 77    Users should not expect this state to be durable, it can be blown away at
 78    at point.
 79
 80    Stored at: ~/.cache/materialize/build_state.toml
 81    """
 82
 83    def __init__(self, path: Path):
 84        self.path = path
 85        if path.is_file():
 86            with open(path) as f:
 87                self.data = toml.load(f)
 88        else:
 89            self.data = {}
 90
 91    @staticmethod
 92    def default_path() -> Path:
 93        home = Path.home()
 94        path = home / ".cache" / "materialize" / "build_state.toml"
 95        return path
 96
 97    @classmethod
 98    def read(cls, namespace: str) -> Any | None:
 99        cache = LocalState(LocalState.default_path())
100        return cache.data.get(namespace, None)
101
102    @classmethod
103    def write(cls, namespace: str, val: Any):
104        cache = LocalState(LocalState.default_path())
105        cache.data[namespace] = val
106
107        Path(os.path.dirname(cache.path)).mkdir(parents=True, exist_ok=True)
108        with open(cache.path, "w+") as f:
109            toml.dump(cache.data, f)
110
111
112class TeleportLocalState:
113    def __init__(self, data: dict[str, Any] | None):
114        self.data = data or {}
115
116    @classmethod
117    def read(cls):
118        return TeleportLocalState(LocalState.read("teleport"))
119
120    def write(self):
121        LocalState.write("teleport", self.data)
122
123    def get_pid(self, app_name: str) -> str | None:
124        existing = self.data.get(app_name, {})
125        return existing.get("pid")
126
127    def set_pid(self, app_name: str, pid: str | None):
128        existing = self.data.get(app_name, {})
129        existing["pid"] = pid
130        self.data[app_name] = existing
131
132    def get_address(self, app_name: str) -> str | None:
133        existing = self.data.get(app_name, {})
134        return existing.get("address")
135
136    def set_address(self, app_name: str, addr: str | None):
137        existing = self.data.get(app_name, {})
138        existing["address"] = addr
139        self.data[app_name] = existing
140
141    def __str__(self):
142        return dedent(
143            f"""
144        TeleportLocalState:
145            data: {self.data}
146        """
147        )