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