Add support for build metadata in version string

ccc/oci.py

@@ -1,3 +1,4 @@
+import collections.abc
 import functools
 import logging
 import traceback
@@ -67,6 +68,8 @@ def oci_client(
     install_logging_handler: bool=True,
     cfg_factory=None,
     http_connection_pool_size:int=16,
+    tag_preprocessing_callback: collections.abc.Callable[[str], str]=None,
+    tag_postprocessing_callback: collections.abc.Callable[[str], str]=None,
 ) -> oc.Client:
     def base_api_lookup(image_reference):
         registry_cfg = model.container_registry.find_config(
@@ -107,6 +110,8 @@ def base_api_lookup(image_reference):
         credentials_lookup=credentials_lookup,
         routes=routes,
         session=session,
+        tag_preprocessing_callback=tag_preprocessing_callback,
+        tag_postprocessing_callback=tag_postprocessing_callback,
     )
 
 

cnudie/util.py

@@ -22,6 +22,8 @@
     | str
 )
 
+META_SEPARATOR = '.build-'
+
 
 def to_component_id(
     component: ComponentId, /
@@ -566,3 +568,24 @@ def enumerate_group_pairs(
             _add_if_not_duplicate(resource_diff.resource_refs_only_right, i)
 
     return resource_diff
+
+
+def sanitise_version(version: str) -> str:
+    '''
+    Additional build metadata as defined in SemVer can be added via `+` to the version. However,
+    OCI registries don't support `+` as tag character, which is why it has to be sanitised, for
+    example using `META_SEPARATOR`.
+    '''
+    sanitised_version = version.replace('+', META_SEPARATOR)
+
+    return sanitised_version
+
+
+def desanitise_version(version: str) -> str:
+    '''
+    This function reverts the sanitisation of the `sanitise_version` function, which allows
+    processing the version the same way as prior to using `sanitise_version`.
+    '''
+    desanitised_version = version.replace(META_SEPARATOR, '+')
+
+    return desanitised_version

oci/client.py

@@ -1,4 +1,5 @@
 import base64
+import collections.abc
 import dataclasses
 import datetime
 import enum
@@ -220,12 +221,19 @@ def blob_url(self, image_reference: typing.Union[str, om.OciImageReference], dig
             digest
         )
 
-    def manifest_url(self, image_reference: typing.Union[str, om.OciImageReference]) -> str:
+    def manifest_url(
+        self,
+        image_reference: typing.Union[str, om.OciImageReference],
+        tag_preprocessing_callback: collections.abc.Callable[[str], str]=None,
+    ) -> str:
         image_reference = om.OciImageReference.to_image_ref(image_reference)
 
         if not (tag := image_reference.tag):
             raise ValueError(f'{image_reference=} does not seem to contain a tag')
 
+        if tag_preprocessing_callback:
+            tag = tag_preprocessing_callback(tag)
+
         return urljoin(
             self.artifact_base_url(image_reference=image_reference),
             'manifests',
@@ -248,10 +256,25 @@ def __init__(
         self,
         credentials_lookup: typing.Callable,
         routes: OciRoutes=OciRoutes(),
-        disable_tls_validation=False,
+        disable_tls_validation: bool=False,
         timeout_seconds: int=None,
         session: requests.Session=None,
+        tag_preprocessing_callback: collections.abc.Callable[[str], str]=None,
+        tag_postprocessing_callback: collections.abc.Callable[[str], str]=None,
     ):
+        '''
+        @param credentials_lookup <Callable>
+        @param routes <OciRoutes>
+        @param disable_tls_validation <bool>
+        @param timeout_seconds <int>
+        @param session <Session>
+        @param tag_preprocessing_callback <Callable>
+            callback which is instrumented _prior_ to interacting with the OCI registry, i.e. useful
+            in case the tag has to be sanitised so it is accepted by the OCI registry
+        @param tag_postprocessing_callback <Callable>
+            callback which is instrumented _after_ interacting with the OCI registry, i.e. useful to
+            revert required sanitisation of `tag_preprocessing_callback`
+        '''
         self.credentials_lookup = credentials_lookup
         self.token_cache = OauthTokenCache()
         if not session:
@@ -260,6 +283,8 @@ def __init__(
             self.session = session
         self.routes = routes
         self.disable_tls_validation = disable_tls_validation
+        self.tag_preprocessing_callback = tag_preprocessing_callback
+        self.tag_postprocessing_callback = tag_postprocessing_callback
 
         if timeout_seconds:
             timeout_seconds = int(timeout_seconds)
@@ -510,7 +535,10 @@ def manifest_raw(
 
         try:
             res = self._request(
-                url=self.routes.manifest_url(image_reference=image_reference),
+                url=self.routes.manifest_url(
+                    image_reference=image_reference,
+                    tag_preprocessing_callback=self.tag_preprocessing_callback,
+                ),
                 image_reference=image_reference,
                 scope=scope,
                 warn_if_not_ok=not absent_ok,
@@ -647,7 +675,10 @@ def head_manifest(
             accept = om.MimeTypes.single_image
 
         res = self._request(
-            url=self.routes.manifest_url(image_reference=image_reference),
+            url=self.routes.manifest_url(
+                image_reference=image_reference,
+                tag_preprocessing_callback=self.tag_preprocessing_callback,
+            ),
             image_reference=image_reference,
             method='HEAD',
             headers={
@@ -706,7 +737,7 @@ def tags(self, image_reference: str):
         # Google-Artifact-Registry (maybe also others) will return http-200 + HTML in certain
         # error cases (e.g. if image_reference contains a pipe (|) character)
         try:
-            return res.json()['tags']
+            tags = res.json()['tags']
         except json.decoder.JSONDecodeError as jde:
             if not (content_type := res.headers['Content-Type']) == 'application/json':
                 jde.add_note(f'unexpected Content-Type: {content_type=}')
@@ -715,6 +746,14 @@ def tags(self, image_reference: str):
 
             raise jde
 
+        if self.tag_postprocessing_callback:
+            tags = [
+                self.tag_postprocessing_callback(tag)
+                for tag in tags
+            ]
+
+        return tags
+
     def has_multiarch(self, image_reference: str) -> bool:
         res = self.head_manifest(
             image_reference=image_reference,
@@ -746,7 +785,10 @@ def put_manifest(
         logger.debug(f'manifest-mimetype: {content_type=}')
 
         res = self._request(
-            url=self.routes.manifest_url(image_reference=image_reference),
+            url=self.routes.manifest_url(
+                image_reference=image_reference,
+                tag_preprocessing_callback=self.tag_preprocessing_callback,
+            ),
             image_reference=image_reference,
             scope=scope,
             method='PUT',
@@ -794,7 +836,10 @@ def delete_manifest(
                 headers = {}
 
             return self._request(
-                url=self.routes.manifest_url(image_reference=image_reference),
+                url=self.routes.manifest_url(
+                    image_reference=image_reference,
+                    tag_preprocessing_callback=self.tag_preprocessing_callback,
+                ),
                 image_reference=image_reference,
                 scope=scope,
                 headers=headers,