shared/data_store/__init__.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+from .connect import connect
+from .get import get
+from .put import put

shared/data_store/connect.py

@@ -0,0 +1,30 @@
+"""Definition of connect function."""
+from __future__ import annotations
+
+import os
+
+from minio import Minio
+
+
+def connect(
+) -> Minio:
+    """Connect to MinIO server"""
+    minio_endpoint = os.getenv('MINIO_ENDPOINT', default=None)
+    assert isinstance(minio_endpoint, str)
+    minio_access_key = os.getenv('MINIO_ACCESS_KEY', default=None)
+    assert isinstance(minio_access_key, str)
+    minio_secret_key = os.getenv('MINIO_SECRET_KEY', default=None)
+    assert isinstance(minio_secret_key, str)
+    minio_bucket_name = os.getenv('MINIO_BUCKET_NAME', default=None)
+    assert isinstance(minio_bucket_name, str)
+    # connect client
+    client = Minio(
+        endpoint=minio_endpoint,
+        access_key=minio_access_key,
+        secret_key=minio_secret_key,
+        secure=False,
+    )
+    # ensure bucket exists
+    if not client.bucket_exists(bucket_name=minio_bucket_name):
+        client.make_bucket(bucket_name=minio_bucket_name)
+    return client

shared/data_store/get.py

@@ -0,0 +1,27 @@
+"""Definition of get function."""
+from __future__ import annotations
+
+import os
+from io import BytesIO
+
+from minio import Minio
+
+
+def get(
+    client: Minio,
+    object_name: str,
+) -> BytesIO:
+    """Get buffer from bucket in MinIO."""
+    assert isinstance(client, Minio)
+    assert isinstance(object_name, str)
+    bucket_name = os.getenv('MINIO_BUCKET_NAME', default=None)
+    assert isinstance(bucket_name, str)
+    # get buffer
+    try:
+        response = client.get_object(bucket_name, object_name)
+        buffer = BytesIO(response.data)
+    finally:
+        response.close()
+        response.release_conn()
+    buffer.seek(0)
+    return buffer

shared/data_store/put.py

@@ -0,0 +1,38 @@
+"""Definition of put function."""
+from __future__ import annotations
+
+import logging
+import os
+from hashlib import md5
+from io import BytesIO
+
+from minio import Minio
+
+
+def put(
+    client: Minio,
+    buffer: BytesIO,
+) -> str:
+    """Put buffer in bucket in MinIO and return MD5 checksum as object name."""
+    assert isinstance(client, Minio)
+    assert isinstance(buffer, BytesIO)
+    bucket_name = os.getenv('MINIO_BUCKET_NAME', default=None)
+    assert isinstance(bucket_name, str)
+    # get md5 of image
+    checksum = md5(buffer.getbuffer()).hexdigest()
+    # prepare for saving
+    num_bytes = buffer.tell()
+    buffer.seek(0)
+    # send data to bucket
+    try:
+        client.put_object(
+            bucket_name=bucket_name,
+            object_name=checksum,
+            length=num_bytes,
+            data=buffer,
+        )
+    except Exception as exc:
+        logging.error('failed saving data to MinIO')
+        raise exc
+    logging.debug('saved data to %s', checksum)
+    return checksum