Add factory for aio_oci-client to ConvenientClientCreation (ccc)

ccc/aio_oci.py

@@ -0,0 +1,84 @@
+import collections.abc
+import functools
+import logging
+
+import aiohttp
+
+import aio_oci.client as oc
+import model.container_registry
+import oci.auth as oa
+
+
+logger = logging.getLogger(__name__)
+
+
+@functools.lru_cache
+def oci_cfg_lookup(
+    cfg_factory=None,
+) -> collections.abc.Callable[[str, oa.Privileges, bool], oa.OciCredentials]:
+    def find_credentials(
+        image_reference: str,
+        privileges: oa.Privileges=oa.Privileges.READONLY,
+        absent_ok: bool=True,
+    ):
+        registry_cfg = model.container_registry.find_config(
+            image_reference=image_reference,
+            privileges=privileges,
+            cfg_factory=cfg_factory,
+        )
+        if not registry_cfg:
+            if absent_ok:
+                return None # fallback to docker-cfg (or try w/o auth)
+            else:
+                raise RuntimeError(f'No credentials found for {image_reference} with {privileges=}'
+            )
+        creds = registry_cfg.credentials()
+        return oa.OciBasicAuthCredentials(
+            username=creds.username(),
+            password=creds.passwd(),
+        )
+
+    return find_credentials
+
+
+@functools.cache
+def oci_client(
+    credentials_lookup: collections.abc.Callable=None,
+    cfg_factory=None,
+    http_connection_pool_size: int | None=None,
+    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(
+            image_reference=image_reference,
+            privileges=None,
+            cfg_factory=cfg_factory,
+        )
+        if registry_cfg and (base_url := registry_cfg.api_base_url()):
+            return base_url
+        return oc.base_api_url(image_reference)
+
+    routes = oc.OciRoutes(base_api_lookup)
+
+    if not credentials_lookup:
+        credentials_lookup = oci_cfg_lookup(cfg_factory=cfg_factory)
+
+    if http_connection_pool_size is None: # 0 is a valid value here (meaning no limitation)
+        connector = aiohttp.TCPConnector()
+    else:
+        connector = aiohttp.TCPConnector(
+            limit=http_connection_pool_size,
+        )
+
+    session = aiohttp.ClientSession(
+        connector=connector,
+    )
+
+    return oc.Client(
+        credentials_lookup=credentials_lookup,
+        routes=routes,
+        session=session,
+        tag_preprocessing_callback=tag_preprocessing_callback,
+        tag_postprocessing_callback=tag_postprocessing_callback,
+    )