added image specific io functions

shared/data_store/get_image.py

@@ -0,0 +1,39 @@
+"""Definition of get_image function."""
+
+import logging
+import os
+from io import BytesIO
+
+from minio import Minio
+from PIL import Image
+
+
+def get_image(
+    client: Minio,
+    object_name: str,
+) -> Image.Image:
+    """Get image from image subfolder in bucket in Minio."""
+    assert isinstance(client, Minio)
+    assert isinstance(object_name, str)
+    # prepare arguments
+    bucket_name = os.getenv('MINIO_BUCKET_NAME')
+    subfolder = 'images'
+    # get object from bucket
+    object_name = f'{subfolder}/{object_name}'
+    try:
+        response = client.get_object(
+            bucket_name=bucket_name,
+            object_name=object_name,
+        )
+        buffer = BytesIO(response.data)
+    except Exception as exc:
+        logging.error('failed getting data from MinIO')
+        raise exc
+    finally:
+        response.close()
+        response.release_conn()
+    buffer.seek(0)
+    # convert data to image
+    image = Image.open(buffer)
+    logging.debug('got data from %s', object_name)
+    return image