"""Definition of get function.""" import logging import os from io import BytesIO from traceback import print_exc 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') # get buffer try: response = client.get_object( bucket_name=bucket_name, object_name=object_name, ) assert response.status == 200 buffer = BytesIO() chunk_size = 2**14 while chunk := response.read(chunk_size): buffer.write(chunk) buffer.seek(0) logging.debug('got %s', object_name) return buffer except Exception as exc: logging.error('failed getting data from MinIO') print_exc() raise exc finally: response.close() response.release_conn()