package_config.py (8981B)
1 #!/usr/bin/env python3 2 # This file is in the public domain. 3 4 """Load, validate, and atomically update Sandcastle package configuration.""" 5 6 from __future__ import annotations 7 8 import json 9 import os 10 import stat 11 import tempfile 12 import tomllib 13 from dataclasses import dataclass 14 from pathlib import Path 15 16 17 class ConfigError(ValueError): 18 pass 19 20 21 SUPPORTED_BUILDERS = {"generic", "pnpm-workspace"} 22 23 24 @dataclass(frozen=True) 25 class RepositoryConfig: 26 name: str 27 url: str 28 builder: str = "generic" 29 30 31 @dataclass(frozen=True) 32 class PackageConfig: 33 name: str 34 repository: str 35 tag: str 36 debian_path: str = "" 37 dependencies: tuple[str, ...] = () 38 enabled: bool = True 39 builder: str | None = None 40 41 42 @dataclass(frozen=True) 43 class PackagingConfig: 44 repositories: dict[str, RepositoryConfig] 45 packages: dict[str, PackageConfig] 46 47 def repository_for(self, package: PackageConfig) -> RepositoryConfig: 48 return self.repositories[package.repository] 49 50 def builder_for(self, package: PackageConfig) -> str: 51 return package.builder or self.repository_for(package).builder 52 53 def enabled_packages(self) -> list[str]: 54 return sorted(name for name, package in self.packages.items() if package.enabled) 55 56 57 def _expect_table(value: object, description: str) -> dict: 58 if not isinstance(value, dict): 59 raise ConfigError(f"{description} must be a table") 60 return value 61 62 63 def _reject_unknown(table: dict, allowed: set[str], description: str) -> None: 64 unknown = sorted(set(table) - allowed) 65 if unknown: 66 raise ConfigError(f"unknown field(s) in {description}: {', '.join(unknown)}") 67 68 69 def _required_string(table: dict, field: str, description: str) -> str: 70 value = table.get(field) 71 if not isinstance(value, str) or not value: 72 raise ConfigError(f"{description}.{field} must be a non-empty string") 73 return value 74 75 76 def _optional_string( 77 table: dict, field: str, default: str, description: str 78 ) -> str: 79 value = table.get(field, default) 80 if not isinstance(value, str): 81 raise ConfigError(f"{description}.{field} must be a string") 82 return value 83 84 85 def parse_config(data: object) -> PackagingConfig: 86 root = _expect_table(data, "configuration") 87 _reject_unknown(root, {"repositories", "packages"}, "configuration") 88 repository_tables = _expect_table(root.get("repositories"), "repositories") 89 package_tables = _expect_table(root.get("packages"), "packages") 90 91 repositories: dict[str, RepositoryConfig] = {} 92 for name, raw_repository in repository_tables.items(): 93 description = f"repositories.{name}" 94 repository = _expect_table(raw_repository, description) 95 _reject_unknown(repository, {"url", "builder"}, description) 96 builder = _optional_string(repository, "builder", "generic", description) 97 if builder not in SUPPORTED_BUILDERS: 98 raise ConfigError(f"{description}.builder is not supported: {builder!r}") 99 repositories[name] = RepositoryConfig( 100 name=name, 101 url=_required_string(repository, "url", description), 102 builder=builder, 103 ) 104 105 packages: dict[str, PackageConfig] = {} 106 for name, raw_package in package_tables.items(): 107 description = f"packages.{name}" 108 package = _expect_table(raw_package, description) 109 _reject_unknown( 110 package, 111 { 112 "repository", 113 "tag", 114 "debian_path", 115 "dependencies", 116 "enabled", 117 "builder", 118 }, 119 description, 120 ) 121 repository = _required_string(package, "repository", description) 122 if repository not in repositories: 123 raise ConfigError(f"{description} references unknown repository {repository!r}") 124 dependencies = package.get("dependencies", []) 125 if not isinstance(dependencies, list) or not all( 126 isinstance(dependency, str) and dependency for dependency in dependencies 127 ): 128 raise ConfigError(f"{description}.dependencies must be an array of strings") 129 if len(set(dependencies)) != len(dependencies): 130 raise ConfigError(f"{description}.dependencies contains duplicates") 131 enabled = package.get("enabled", True) 132 if not isinstance(enabled, bool): 133 raise ConfigError(f"{description}.enabled must be a boolean") 134 builder = package.get("builder") 135 if builder is not None and (not isinstance(builder, str) or not builder): 136 raise ConfigError(f"{description}.builder must be a non-empty string") 137 if builder is not None and builder not in SUPPORTED_BUILDERS: 138 raise ConfigError(f"{description}.builder is not supported: {builder!r}") 139 packages[name] = PackageConfig( 140 name=name, 141 repository=repository, 142 tag=_required_string(package, "tag", description), 143 debian_path=_optional_string(package, "debian_path", "", description), 144 dependencies=tuple(dependencies), 145 enabled=enabled, 146 builder=builder, 147 ) 148 149 for package in packages.values(): 150 for dependency in package.dependencies: 151 if dependency not in packages: 152 raise ConfigError( 153 f"packages.{package.name} references unknown dependency {dependency!r}" 154 ) 155 _check_dependency_cycles(packages) 156 return PackagingConfig(repositories=repositories, packages=packages) 157 158 159 def _check_dependency_cycles(packages: dict[str, PackageConfig]) -> None: 160 permanent: set[str] = set() 161 temporary: list[str] = [] 162 163 def visit(name: str) -> None: 164 if name in permanent: 165 return 166 if name in temporary: 167 cycle = temporary[temporary.index(name) :] + [name] 168 raise ConfigError(f"package dependency cycle: {' -> '.join(cycle)}") 169 temporary.append(name) 170 for dependency in packages[name].dependencies: 171 visit(dependency) 172 temporary.pop() 173 permanent.add(name) 174 175 for name in sorted(packages): 176 visit(name) 177 178 179 def load_config(path: str | Path) -> PackagingConfig: 180 path = Path(path) 181 try: 182 with path.open("rb") as config_file: 183 return parse_config(tomllib.load(config_file)) 184 except (OSError, tomllib.TOMLDecodeError) as exc: 185 raise ConfigError(f"unable to read {path}: {exc}") from exc 186 187 188 def _toml_string(value: str) -> str: 189 return json.dumps(value, ensure_ascii=False) 190 191 192 def _toml_string_array(values: tuple[str, ...]) -> str: 193 return "[" + ", ".join(_toml_string(value) for value in values) + "]" 194 195 196 def serialize_config(config: PackagingConfig) -> str: 197 lines = [ 198 "# Package definitions for Sandcastle.", 199 "# This file is rewritten in canonical form by `sandcastle-upgrade`.", 200 "", 201 ] 202 for name in sorted(config.repositories): 203 repository = config.repositories[name] 204 lines.extend( 205 [ 206 f"[repositories.{_toml_string(name)}]", 207 f"url = {_toml_string(repository.url)}", 208 f"builder = {_toml_string(repository.builder)}", 209 "", 210 ] 211 ) 212 for name in sorted(config.packages): 213 package = config.packages[name] 214 lines.extend( 215 [ 216 f"[packages.{_toml_string(name)}]", 217 f"repository = {_toml_string(package.repository)}", 218 f"tag = {_toml_string(package.tag)}", 219 f"debian_path = {_toml_string(package.debian_path)}", 220 f"dependencies = {_toml_string_array(package.dependencies)}", 221 f"enabled = {'true' if package.enabled else 'false'}", 222 ] 223 ) 224 if package.builder is not None: 225 lines.append(f"builder = {_toml_string(package.builder)}") 226 lines.append("") 227 return "\n".join(lines) 228 229 230 def write_config(path: str | Path, config: PackagingConfig) -> None: 231 path = Path(path) 232 serialized = serialize_config(config) 233 reparsed = parse_config(tomllib.loads(serialized)) 234 if reparsed != config: 235 raise ConfigError("serialized configuration did not round-trip") 236 237 mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else 0o644 238 fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) 239 try: 240 os.fchmod(fd, mode) 241 with os.fdopen(fd, "w", encoding="utf-8") as config_file: 242 config_file.write(serialized) 243 config_file.flush() 244 os.fsync(config_file.fileno()) 245 os.replace(temporary_name, path) 246 directory_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) 247 try: 248 os.fsync(directory_fd) 249 finally: 250 os.close(directory_fd) 251 except BaseException: 252 try: 253 os.unlink(temporary_name) 254 except FileNotFoundError: 255 pass 256 raise