Expose parsed docker images

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 5.4.3
+current_version = 5.4.4
 commit = True
 tag = False
 

.github/workflows/docker.yaml

@@ -5,7 +5,7 @@ on:
       - main
 
 env:
-  VERSION: 5.4.3
+  VERSION: 5.4.4
 
 permissions:
   contents: read

cpg_utils/cloud.py

@@ -7,13 +7,15 @@
 import traceback
 import urllib.parse
 from collections import defaultdict
+from datetime import datetime
 from typing import Any, NamedTuple
 
 # pylint: disable=no-name-in-module
 import google.api_core.exceptions
 import google.auth.transport
 import google.oauth2
 from deprecated import deprecated
+from google.api_core.datetime_helpers import to_milliseconds
 from google.auth import (
     credentials as google_auth_credentials,
 )
@@ -31,6 +33,7 @@
 from google.cloud import artifactregistry, secretmanager
 from google.oauth2 import credentials as oauth2_credentials
 from google.oauth2 import service_account
+from zoneinfo import ZoneInfo
 
 _CLOUD_SDK_MISSING_CREDENTIALS = """\
 Your default credentials were not found. To set up Application Default Credentials, \
@@ -43,6 +46,8 @@
     _EXTERNAL_ACCOUNT_TYPE,
 )
 
+TZ = ZoneInfo('Australia/Sydney')
+
 
 def email_from_id_token(id_token_jwt: str) -> str:
     """Decodes the ID token (JWT) to get the email address of the caller.
@@ -129,8 +134,8 @@ class DockerImage(NamedTuple):
     name: str
     uri: str
     tag_uri: str
-    size: str
-    build_time: str
+    size: int
+    build_time: datetime
 
 
 _repo_image_tags: dict[str, defaultdict[str, dict[str, DockerImage]]] = {}
@@ -151,19 +156,43 @@ def _ensure_image_tags_loaded(project: str, location: str, repository: str) -> N
         name_and_checksum = image.name.rpartition('/dockerImages/')[2]
         name = urllib.parse.unquote(name_and_checksum).rpartition('@')[0]
         base_uri = image.uri.rpartition('@')[0]
+
+        # digest the Google proto DatetimeWithNanoseconds object into a standard python datetime
+        images_build_time_ms = to_milliseconds(image.build_time)
+        build_datetime = datetime.fromtimestamp(images_build_time_ms // 1000, tz=TZ)
+
         for tag in image.tags:
             image_tags[name][tag] = DockerImage(
                 image.name,
                 image.uri,
                 f'{base_uri}:{tag}',
                 image.image_size_bytes,
-                image.build_time,
+                build_datetime,
             )
 
     image_tags.default_factory = None
     _repo_image_tags[repository] = image_tags
 
 
+def get_tags_for_image(
+    project: str,
+    location: str,
+    repository: str,
+    image: str,
+) -> dict[str, DockerImage]:
+    """
+    Ensure available images are loaded, then request all available tags from a specific image
+    """
+    _ensure_image_tags_loaded(project, location, repository)
+    if repository not in _repo_image_tags:
+        raise ValueError(
+            f'repository "{repository}" is not available in the collected tags.',
+        )
+    if image not in _repo_image_tags[repository]:
+        raise ValueError(f'image "{image}" is not available in the collected tags.')
+    return _repo_image_tags[repository][image]
+
+
 def find_image(repository: str | None, name: str, version: str) -> DockerImage:
     """Returns image details or raises ValueError if the image or tag does not exist."""
     repository = f'images-{repository}' if repository is not None else 'images'

setup.py

@@ -8,7 +8,7 @@
 setup(
     name='cpg-utils',
     # This tag is automatically updated by bumpversion
-    version='5.4.3',
+    version='5.4.4',
     description='Library of convenience functions specific to the CPG',
     long_description=long_description,
     long_description_content_type='text/markdown',