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