updatebot.py

@@ -5,12 +5,12 @@ import logging
 import functools
 import os
 import re
-import subprocess
 import tempfile
 import threading
+import tomllib
 import urllib.parse
 from pathlib import Path
-from typing import ClassVar, Iterable, Literal, Union, Optional
+from typing import ClassVar, Literal, Union, Optional
 
 import colorlog
 import pydantic
@@ -112,33 +112,19 @@ Source = Union[
 ]
 
 
-class ImageDef(pydantic.BaseModel):
+class BaseProject(abc.ABC, pydantic.BaseModel):
-    name: str
+    path: Optional[Path] = None
-    image: str
     source: Source
+    image: str
     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_updates(self, basedir: Path) -> Iterable[tuple[ImageDef, str]]:
+    def apply_update(self, path: Path, version: str) -> None:
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def resource_diff(self, basedir: Path) -> Optional[str]:
         raise NotImplementedError
 
 
 class KustomizeProject(BaseProject):
     kind: Literal['kustomize']
-    images: list[ImageDef]
 
     kustomize_files: ClassVar[list[str]] = [
         'kustomization.yaml',
@@ -146,60 +132,41 @@ class KustomizeProject(BaseProject):
         'Kustomization',
     ]
 
-    def apply_updates(self, basedir: Path) -> Iterable[tuple[ImageDef, str]]:
+    def apply_update(self, path: Path, version: str) -> None:
-        path = basedir / self.path
         for filename in self.kustomize_files:
             filepath = path / filename
             if filepath.is_file():
                 break
         else:
-            raise ValueError(
+            filepath = path / self.kustomize_files[0]
-                f'Could not find Kustomize config for {self.name}'
-            )
-
-        for image in self.images:
         yaml = ruamel.yaml.YAML()
         with filepath.open('rb') as f:
             kustomization = yaml.load(f)
         images = kustomization.setdefault('images', [])
-            version = image.source.get_latest_version()
+        new_tag = self.tag_format.format(version=version)
-            new_tag = image.tag_format.format(version=version)
+        for image in images:
-            for i in images:
+            if image['name'] == self.image:
-                if i['name'] == image.image:
+                image['newTag'] = new_tag
-                    i['newTag'] = new_tag
                 break
         else:
-                images.append({'name': image.image, 'newTag': new_tag})
+            images.append({'name': self.image, 'newTag': new_tag})
         with filepath.open('wb') as f:
             yaml.dump(kustomization, f)
-            yield (image, version)
-
-    def resource_diff(self, basedir: Path) -> Optional[str]:
-        path = basedir / self.path
-        cmd = ['kubectl', 'diff', '-k', path]
-        try:
-            p = subprocess.run(
-                cmd,
-                stdin=subprocess.DEVNULL,
-                capture_output=True,
-                check=False,
-                encoding='utf-8',
-            )
-        except FileNotFoundError as e:
-            log.warning('Cannot generate resource diff: %s', e)
-            return None
-        if p.returncode != 0 and not p.stdout:
-            log.error('Failed to generate resource diff: %s', p.stderr)
-            return None
-        return p.stdout
 
 
-Project = KustomizeProject
+class DirectoryProject(BaseProject):
+    kind: Literal['dir'] | Literal['directory']
+
+
+Project = Union[
+    KustomizeProject,
+    DirectoryProject,
+]
 
 
 class RepoConfig(pydantic.BaseModel):
     url: str
-    token_file: Optional[Path] = None
+    token_file: Path
     branch: str = 'master'
 
     @functools.cached_property
@@ -209,18 +176,16 @@ class RepoConfig(pydantic.BaseModel):
         return urllib.parse.urlunsplit(urlparts)
 
     @functools.cached_property
-    def auth_token(self) -> Optional[str]:
+    def auth_token(self) -> str:
-        if self.token_file:
         return self.token_file.read_text().strip()
 
     def get_git_url(self) -> str:
         session = _get_session()
-        headers = {}
-        if token := self.auth_token:
-            headers['Authorization'] = f'Bearer {token}'
         r = session.get(
             self.repo_api_url,
-            headers=headers,
+            headers={
+                'Authorization': f'token {self.auth_token}',
+            },
         )
         data = r.json()
         if ssh_url := data.get('ssh_url'):
@@ -228,11 +193,7 @@ class RepoConfig(pydantic.BaseModel):
         return data['clone_url']
 
     def create_pr(
-        self,
+        self, title: str, source_branch: str, target_branch: str
-        title: str,
-        source_branch: str,
-        target_branch: str,
-        body: Optional[str] = None,
     ) -> None:
         session = _get_session()
         r = session.post(
@@ -244,7 +205,6 @@ class RepoConfig(pydantic.BaseModel):
                 'title': title,
                 'base': target_branch,
                 'head': source_branch,
-                'body': body,
             },
         )
         log.log(TRACE, '%r', r.content)
@@ -263,7 +223,7 @@ class RepoConfig(pydantic.BaseModel):
 
 class Config(pydantic.BaseModel):
     repo: RepoConfig
-    projects: list[Project]
+    projects: dict[str, Project]
 
 
 class Arguments:
@@ -274,26 +234,24 @@ class Arguments:
 
 
 def update_project(
-    repo: git.Repo, project: Project
+    repo: git.Repo, name: str, project: Project
-) -> tuple[Optional[str], Optional[str]]:
+) -> Optional[git.Commit]:
     basedir = Path(repo.working_dir)
-    title = None
+    log.debug('Checking for latest version of %s', name)
-    for image, version in project.apply_updates(basedir):
+    latest = project.source.get_latest_version()
-        log.info('Updating %s to %s', image.name, version)
+    log.info('Found version %s for %s', latest, name)
+    log.debug('Applying update for %s version %s', name, latest)
+    path = basedir / (project.path or name)
+    project.apply_update(path, latest)
     if repo.index.diff(None):
-            log.debug('Committing changes to %s', project.path)
+        log.debug('Committing changes to %s', path)
-            repo.index.add(str(project.path))
+        repo.index.add(str(path))
-            c = repo.index.commit(f'{image.name}: Update to {version}')
+        c = repo.index.commit(f'{name}: Update to {latest}')
         log.info('Commited %s %s', str(c)[:7], c.summary)
-            if not title:
+        return c
-                if not isinstance(c.summary, str):
-                    title = bytes(c.summary).decode('utf-8')
-                else:
-                    title = c.summary
     else:
         log.info('No changes to commit')
-    diff = project.resource_diff(basedir)
+        return None
-    return title, diff
 
 
 def parse_args() -> Arguments:
@@ -336,12 +294,23 @@ def setup_logging() -> None:
 def main() -> None:
     setup_logging()
     args = parse_args()
-    yaml = ruamel.yaml.YAML()
     with args.config.open('rb') as f:
-        data = yaml.load(f)
+        data = tomllib.load(f)
     config = Config.model_validate(data)
     log.debug('Using configuration: %s', config)
-    projects = args.projects or [p.name for p in config.projects]
+    all_projects = list(config.projects.keys())
+    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:
@@ -350,43 +319,29 @@ 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)
-        for project in config.projects:
         log.debug('Checking out new branch: %s', args.branch_name)
         repo.heads[0].checkout(force=True, B=args.branch_name)
         title = None
-            description = None
+        for project in projects:
-            if project.name not in projects:
+            commit = update_project(repo, project, config.projects[project])
-                continue
+            if commit and not title:
-            title, diff = update_project(repo, project)
+                if not isinstance(commit.summary, str):
+                    title = bytes(commit.summary).decode(
+                        'utf-8', errors='replace'
+                    )
+                else:
+                    title = commit.summary
         if not title:
             log.info('No changes made')
-                continue
+            return
-            if diff:
-                description = (
-                    '<details>\n<summary>Resource diff</summary>\n\n'
-                    f'```diff\n{diff}```\n'
-                    '</details>'
-                )
-            if not args.dry_run:
         repo.head.reference.set_tracking_branch(
             git.RemoteReference(
                 repo, f'refs/remotes/origin/{args.branch_name}'
             )
         )
+        if not args.dry_run:
             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,
-                    description,
-                )
-            else:
-                print(
-                    'Would create PR',
-                    f'{args.branch_name} → {config.repo.branch}:',
-                    title,
-                )
-                print(description or '')
 
 
 if __name__ == '__main__':