tools: add releases.json TypedDict wrapper

Avoid duplicated code for reading/writing releases.json, and ensure the
type checker has information about its contents.

Author: bgilbert

tools/create_release.py

@@ -26,7 +26,7 @@
 import json
 
 from pathlib import Path
-from utils import is_ci, is_debianlike, read_wrap, write_wrap
+from utils import Releases, is_ci, is_debianlike, read_wrap, write_wrap
 
 class CreateRelease:
     def __init__(self, repo: T.Optional[str], token: T.Optional[str], tag: str):
@@ -155,8 +155,7 @@ def create_source_fallback(self) -> None:
         self.wrap_section['source_fallback_url'] = f'https://github.com/mesonbuild/wrapdb/releases/download/{self.tag}/{filename.name}'
 
 def run(repo: T.Optional[str], token: T.Optional[str]) -> None:
-    with open('releases.json', 'r') as f:
-        releases = json.load(f)
+    releases = Releases.load()
     stdout = subprocess.check_output(['git', 'tag'])
     tags = [t.strip() for t in stdout.decode().splitlines()]
     for name, info in releases.items():

tools/format.py

@@ -15,18 +15,15 @@
 # limitations under the License.
 
 from __future__ import annotations
-import json
 from pathlib import Path
 import subprocess
 
-from utils import format_meson, read_wrap
+from utils import Releases, format_meson, read_wrap
 
 FORMAT_FILES = {'meson.build', 'meson_options.txt', 'meson.options'}
 
 def main() -> None:
-    with open('releases.json', 'r', encoding='utf-8') as f:
-        releases = json.load(f)
-
+    releases = Releases.load()
     tags = set(subprocess.check_output(['git', 'tag', '--merged'], text=True).splitlines())
 
     files = []

tools/internalize_sources.py

@@ -17,16 +17,14 @@
 from __future__ import annotations
 from argparse import ArgumentParser
 from hashlib import sha256
-import json
 from pathlib import Path
 import subprocess
 
-from utils import read_wrap, write_wrap, wrap_path
+from utils import Releases, read_wrap, write_wrap, wrap_path
 
 class Internalizer:
     def __init__(self, all=False):
-        with open('releases.json') as fh:
-            releases = json.load(fh)
+        releases = Releases.load()
         tags = set(
             subprocess.check_output(['git', 'tag'], text=True).splitlines()
         )

tools/sanity_checks.py

@@ -29,7 +29,7 @@
 import shutil
 
 from pathlib import Path
-from utils import Version, ci_group, is_ci, is_alpinelike, is_debianlike, is_macos, is_windows, is_msys, read_wrap, FormattingError, format_meson
+from utils import Releases, Version, ci_group, is_ci, is_alpinelike, is_debianlike, is_macos, is_windows, is_msys, read_wrap, FormattingError, format_meson
 
 PERMITTED_FILES = {'generator.sh', 'meson.build', 'meson_options.txt', 'meson.options', 'LICENSE.build'}
 PER_PROJECT_PERMITTED_FILES: dict[str, set[str]] = {
@@ -181,19 +181,13 @@ class CiConfigProject(T.TypedDict, total=False):
         skip_tests: bool
 
 
-    class ReleasesProject(T.TypedDict, total=False):
-        dependency_names: list[str]
-        program_names: list[str]
-        versions: T.Required[list[str]]
-
-
 class TestReleases(unittest.TestCase):
     # requires casts for special keys e.g. broken_*
     ci_config: dict[str, CiConfigProject]
     fatal_warnings: bool
     annotate_context: bool
     skip_build: bool
-    releases: dict[str, ReleasesProject]
+    releases: Releases
     skip: list[str]
     tags: set[str]
     timeout_multiplier: float
@@ -216,14 +210,12 @@ def setUpClass(cls):
             print(f'Ignoring unreachable tags: {stdout.decode().splitlines()}')
 
         try:
-            fn = 'releases.json'
-            with open(fn, 'r', encoding='utf-8') as f:
-                cls.releases = json.load(f)
+            cls.releases = Releases.load()
             fn = 'ci_config.json'
             with open(fn, 'r', encoding='utf-8') as f:
                 cls.ci_config = json.load(f)
-        except json.decoder.JSONDecodeError:
-            raise RuntimeError(f'file {fn} is malformed')
+        except json.decoder.JSONDecodeError as ex:
+            raise RuntimeError('metadata is malformed') from ex
 
         system = platform.system().lower()
         cls.skip = T.cast(T.List[str], cls.ci_config[f'broken_{system}'])

tools/utils.py

@@ -13,10 +13,12 @@
 # limitations under the License.
 
 from __future__ import annotations
+import abc
 import configparser
 from contextlib import contextmanager
 import functools
 import io
+import json
 import operator
 import os
 from pathlib import Path
@@ -106,6 +108,32 @@ def __cmp(self, other: 'Version', comparator: T.Callable[[T.Any, T.Any], bool])
         # versions are equal, so compare revisions
         return comparator(self._r, other._r)
 
+class _JSONFile(abc.ABC):
+    FILENAME: str
+
+    def __init__(self, _: T.Any): ...
+
+    @classmethod
+    def load(cls) -> T.Self:
+        with open(cls.FILENAME, encoding='utf-8') as f:
+            return cls(json.load(f))
+
+    def encode(self) -> str:
+        return json.dumps(self, indent=2, sort_keys=True) + '\n'
+
+    def save(self) -> None:
+        with open(f'{self.FILENAME}.new', 'w', encoding='utf-8') as f:
+            f.write(self.encode())
+        os.rename(f'{self.FILENAME}.new', self.FILENAME)
+
+class ProjectReleases(T.TypedDict):
+    dependency_names: T.NotRequired[list[str]]
+    program_names: T.NotRequired[list[str]]
+    versions: list[str]
+
+class Releases(dict[str, ProjectReleases], _JSONFile):
+    FILENAME = 'releases.json'
+
 def wrap_path(name: str) -> Path:
     return Path('subprojects', f'{name}.wrap')
 

tools/versions.py

@@ -30,7 +30,7 @@
 
 import requests
 
-from utils import read_wrap, wrap_path
+from utils import Releases, read_wrap, wrap_path
 
 WRAP_URL_TEMPLATE = (
     'https://github.com/mesonbuild/wrapdb/blob/master/subprojects/{0}.wrap'
@@ -60,12 +60,6 @@ class AnityaPackage(T.TypedDict):
     version: str
 
 
-class WrapInfo(T.TypedDict):
-    versions: list[str]
-    dependency_names: list[str]
-    program_names: list[str]
-
-
 @cache
 def get_upstream_versions() -> dict[str, str]:
     '''Query Anitya and return a dict: wrap_name -> upstream_version.'''
@@ -107,32 +101,27 @@ def sub(name: str, old: str, new: str) -> None:
 
 
 @cache
-def get_releases(commit=None) -> dict[str, WrapInfo]:
-    '''Parse and return releases.json for the specified commit or the working
-    tree.'''
-    if commit is not None:
-        data = subprocess.check_output(
-            ['git', 'cat-file', 'blob', f'{commit}:releases.json'], text=True
-        )
-    else:
-        with open('releases.json') as f:
-            data = f.read()
-    return json.loads(data)
+def get_commit_releases(commit: str) -> Releases:
+    '''Parse and return releases.json for the specified commit.'''
+    data = subprocess.check_output(
+        ['git', 'cat-file', 'blob', f'{commit}:releases.json'], text=True
+    )
+    return Releases(json.loads(data))
 
 
 def get_wrap_versions() -> dict[str, str]:
     '''Return a dict: wrap_name -> wrapdb_version.'''
     return {
         name: info['versions'][0].split('-')[0]
-        for name, info in get_releases().items()
+        for name, info in Releases.load().items()
         if name not in DEPRECATED_WRAPS
     }
 
 
 def get_port_wraps() -> set[str]:
     '''Return the names of wraps that have a patch directory.'''
     ports = set()
-    for name, info in get_releases().items():
+    for name, info in Releases.load().items():
         wrap = read_wrap(name)
         if wrap.has_option('wrap-file', 'patch_directory'):
             ports.add(name)
@@ -190,17 +179,9 @@ def update_wrap(name: str, old_ver: str, new_ver: str) -> None:
         f.write(''.join(lines))
 
 
-def write_releases(releases: dict[str, WrapInfo]) -> None:
-    '''Write modified releases.json.'''
-    with open('releases.json.new', 'w') as f:
-        json.dump(releases, f, indent=2, sort_keys=True)
-        f.write('\n')
-    os.rename('releases.json.new', 'releases.json')
-
-
 def update_revisions(args: Namespace) -> None:
     # run queries
-    releases = get_releases()
+    releases = Releases.load()
     cur_vers = get_wrap_versions()
 
     # decide what to update
@@ -217,7 +198,7 @@ def update_revisions(args: Namespace) -> None:
         print(f'Updating {name} revision...')
         cur_rev = int(releases[name]['versions'][0].split('-')[1])
         releases[name]['versions'].insert(0, f'{cur_vers[name]}-{cur_rev + 1}')
-    write_releases(releases)
+    releases.save()
 
 
 def do_autoupdate(args: Namespace) -> None:
@@ -226,7 +207,7 @@ def do_autoupdate(args: Namespace) -> None:
         return update_revisions(args)
 
     # run queries
-    releases = get_releases()
+    releases = Releases.load()
     cur_vers = get_wrap_versions()
     upstream_vers = get_upstream_versions()
     ports = get_port_wraps()
@@ -257,7 +238,7 @@ def do_autoupdate(args: Namespace) -> None:
                     print(f'Updating {name}...')
                 update_wrap(name, cur_ver, upstream_ver)
                 releases[name]['versions'].insert(0, f'{upstream_ver}-1')
-                write_releases(releases)
+                releases.save()
         except Exception as e:
             print(e, file=sys.stderr)
             failures += 1
@@ -266,8 +247,8 @@ def do_autoupdate(args: Namespace) -> None:
 
 
 def do_commit(args: Namespace) -> None:
-    old_releases = get_releases('HEAD')
-    new_releases = get_releases()
+    old_releases = get_commit_releases('HEAD')
+    new_releases = Releases.load()
 
     # we don't validate any invariants checked by sanity_checks.py
     changed_wraps = [