moved functions to shared folder

model/src/data_store/__init__.py

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

model/src/data_store/get.py

@@ -1,27 +0,0 @@
-"""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

model/src/data_store/put.py

@@ -1,30 +0,0 @@
-"""Definition of put function."""
-from __future__ import annotations
-
-import os
-from io import BytesIO
-
-from minio import Minio
-
-
-def put(
-    client: Minio,
-    buffer: BytesIO,
-    object_name: str,
-) -> None:
-    """Put buffer in bucket in MinIO."""
-    assert isinstance(client, Minio)
-    assert isinstance(buffer, BytesIO)
-    assert isinstance(object_name, str)
-    bucket_name = os.getenv('MINIO_BUCKET_NAME', default=None)
-    assert isinstance(bucket_name, str)
-    # prepare for saving
-    num_bytes = buffer.tell()
-    buffer.seek(0)
-    # send data to bucket
-    client.put_object(
-        bucket_name=bucket_name,
-        object_name=object_name,
-        length=num_bytes,
-        data=buffer,
-    )

shared/data_store/__init__.py

@@ -1,6 +1,6 @@
-from __future__ import annotations
-
 from .connect import connect
 from .delete import delete
 from .get import get
+from .get_image import get_image
 from .put import put
+from .put_image import put_image

shared/data_store/get.py

@@ -1,5 +1,4 @@
 """Definition of get function."""
-from __future__ import annotations
 
 import os
 from io import BytesIO
@@ -14,8 +13,7 @@ def get(
     """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)
+    bucket_name = os.getenv('MINIO_BUCKET_NAME')
     # get buffer
     try:
         response = client.get_object(bucket_name, object_name)