Rework data model to group images into projects

A "project" now refers to an application deployed into Kubernetes, which
includes one or more "images."  This is really the grouping I wanted in
the beginning, which I tried to achieve using separate configuration
files.  Unfortunately, this made the original "projects" too
independent, making it difficult to produce the manifest diff I wanted
to add to the PR descriptions.  It was also cumbersome managing multiple
config files and therefore multiple CronJobs in Kubernetes.

The new data model is a lot deeper than the original one, making TOML a
lot less nice.  YAML definitely handles nested data structures better,
despite its shortcomings.  Having to repeat nested table names in TOML
is quite cumbersome.

updatebot.py

@@ -7,10 +7,9 @@ import os
 import re
 import tempfile
 import threading
-import tomllib
 import urllib.parse
 from pathlib import Path
-from typing import ClassVar, Literal, Union, Optional
+from typing import ClassVar, Iterable, Literal, Union, Optional
 
 import colorlog
 import pydantic
@@ -112,19 +111,29 @@ Source = Union[
 ]
 
 
-class BaseProject(abc.ABC, pydantic.BaseModel):
+class ImageDef(pydantic.BaseModel):
-    path: Optional[Path] = None
+    name: str
-    source: Source
     image: str
+    source: Source
     tag_format: str = '{version}'
 
+
+class BaseProject(abc.ABC, pydantic.BaseModel):
+    name: str
+    path: Path
+
+    def __init__(self, **data) -> None:
+        data.setdefault('path', data.get('name'))
+        super().__init__(**data)
+
     @abc.abstractmethod
-    def apply_update(self, path: Path, version: str) -> None:
+    def apply_updates(self, basedir: Path) -> Iterable[tuple[ImageDef, str]]:
         raise NotImplementedError
 
 
 class KustomizeProject(BaseProject):
     kind: Literal['kustomize']
+    images: list[ImageDef]
 
     kustomize_files: ClassVar[list[str]] = [
         'kustomization.yaml',
@@ -132,36 +141,36 @@ class KustomizeProject(BaseProject):
         'Kustomization',
     ]
 
-    def apply_update(self, path: Path, version: str) -> None:
+    def apply_updates(self, basedir: Path) -> Iterable[tuple[ImageDef, str]]:
+        path = basedir / self.path
         for filename in self.kustomize_files:
             filepath = path / filename
             if filepath.is_file():
                 break
         else:
-            filepath = path / self.kustomize_files[0]
+            raise ValueError(
-        yaml = ruamel.yaml.YAML()
+                f'Could not find Kustomize config for {self.name}'
-        with filepath.open('rb') as f:
+            )
-            kustomization = yaml.load(f)
+
-        images = kustomization.setdefault('images', [])
+        for image in self.images:
-        new_tag = self.tag_format.format(version=version)
+            yaml = ruamel.yaml.YAML()
-        for image in images:
+            with filepath.open('rb') as f:
-            if image['name'] == self.image:
+                kustomization = yaml.load(f)
-                image['newTag'] = new_tag
+            images = kustomization.setdefault('images', [])
-                break
+            version = image.source.get_latest_version()
-        else:
+            new_tag = image.tag_format.format(version=version)
-            images.append({'name': self.image, 'newTag': new_tag})
+            for i in images:
-        with filepath.open('wb') as f:
+                if i['name'] == image.image:
-            yaml.dump(kustomization, f)
+                    i['newTag'] = new_tag
+                    break
+            else:
+                images.append({'name': image.image, 'newTag': new_tag})
+            with filepath.open('wb') as f:
+                yaml.dump(kustomization, f)
+            yield (image, version)
 
 
-class DirectoryProject(BaseProject):
+Project = KustomizeProject
-    kind: Literal['dir'] | Literal['directory']
-
-
-Project = Union[
-    KustomizeProject,
-    DirectoryProject,
-]
 
 
 class RepoConfig(pydantic.BaseModel):
@@ -225,7 +234,7 @@ class RepoConfig(pydantic.BaseModel):
 
 class Config(pydantic.BaseModel):
     repo: RepoConfig
-    projects: dict[str, Project]
+    projects: list[Project]
 
 
 class Arguments:
@@ -235,25 +244,24 @@ class Arguments:
     projects: list[str]
 
 
-def update_project(
+def update_project(repo: git.Repo, project: Project) -> Optional[str]:
-    repo: git.Repo, name: str, project: Project
-) -> Optional[git.Commit]:
     basedir = Path(repo.working_dir)
-    log.debug('Checking for latest version of %s', name)
+    title = None
-    latest = project.source.get_latest_version()
+    for (image, version) in project.apply_updates(basedir):
-    log.info('Found version %s for %s', latest, name)
+        log.info('Updating %s to %s', image.name, version)
-    log.debug('Applying update for %s version %s', name, latest)
+        if repo.index.diff(None):
-    path = basedir / (project.path or name)
+            log.debug('Committing changes to %s', project.path)
-    project.apply_update(path, latest)
+            repo.index.add(str(project.path))
-    if repo.index.diff(None):
+            c = repo.index.commit(f'{image.name}: Update to {version}')
-        log.debug('Committing changes to %s', path)
+            log.info('Commited %s %s', str(c)[:7], c.summary)
-        repo.index.add(str(path))
+            if not title:
-        c = repo.index.commit(f'{name}: Update to {latest}')
+                if not isinstance(c.summary, str):
-        log.info('Commited %s %s', str(c)[:7], c.summary)
+                    title = bytes(c.summary).decode('utf-8')
-        return c
+                else:
-    else:
+                    title = c.summary
-        log.info('No changes to commit')
+        else:
-        return None
+            log.info('No changes to commit')
+    return title
 
 
 def parse_args() -> Arguments:
@@ -296,23 +304,12 @@ def setup_logging() -> None:
 def main() -> None:
     setup_logging()
     args = parse_args()
+    yaml = ruamel.yaml.YAML()
     with args.config.open('rb') as f:
-        data = tomllib.load(f)
+        data = yaml.load(f)
     config = Config.model_validate(data)
     log.debug('Using configuration: %s', config)
-    all_projects = list(config.projects.keys())
+    projects = args.projects or [p.name for p in config.projects]
-    projects = []
-    if args.projects:
-        for project in args.projects:
-            if project not in all_projects:
-                log.warning('Unknown project: %s', project)
-                continue
-            projects.append(project)
-    else:
-        projects = all_projects
-    if not projects:
-        log.error('No projects')
-        raise SystemExit(1)
     if log.isEnabledFor(logging.INFO):
         log.info('Updating projects: %s', ', '.join(projects))
     with tempfile.TemporaryDirectory(prefix='updatebot.') as d:
@@ -321,29 +318,33 @@ def main() -> None:
         log.debug('Retreiving repository Git URL')
         repo_url = config.repo.get_git_url()
         repo = git.Repo.clone_from(repo_url, d, depth=1, b=config.repo.branch)
-        log.debug('Checking out new branch: %s', args.branch_name)
+        for project in config.projects:
-        repo.heads[0].checkout(force=True, B=args.branch_name)
+            log.debug('Checking out new branch: %s', args.branch_name)
-        title = None
+            repo.heads[0].checkout(force=True, B=args.branch_name)
-        for project in projects:
+            title = None
-            commit = update_project(repo, project, config.projects[project])
+            if project.name not in projects:
-            if commit and not title:
+                continue
-                if not isinstance(commit.summary, str):
+            title = update_project(repo, project)
-                    title = bytes(commit.summary).decode(
+            if not title:
-                        'utf-8', errors='replace'
+                log.info('No changes made')
-                    )
+                continue
-                else:
+            repo.head.reference.set_tracking_branch(
-                    title = commit.summary
+                git.RemoteReference(
-        if not title:
+                    repo, f'refs/remotes/origin/{args.branch_name}'
-            log.info('No changes made')
+                )
-            return
-        repo.head.reference.set_tracking_branch(
-            git.RemoteReference(
-                repo, f'refs/remotes/origin/{args.branch_name}'
             )
-        )
+            if not args.dry_run:
-        if not args.dry_run:
+                repo.remote().push(force=True)
-            repo.remote().push(force=True)
+                config.repo.create_pr(
-            config.repo.create_pr(title, args.branch_name, config.repo.branch)
+                    title, args.branch_name, config.repo.branch
+                )
+            else:
+                log.info(
+                    'Would create PR %s → %s: %s',
+                    config.repo.branch,
+                    args.branch_name,
+                    title,
+                )
 
 
 if __name__ == '__main__':