import io from minio.error import S3Error class MinioClient: def __init__(self, client, bucket: str): self._client = client self._bucket = bucket def list_objects(self, prefix: str) -> list[dict]: objects = self._client.list_objects(self._bucket, prefix=prefix, recursive=False) result = [] for obj in objects: result.append({ "key": obj.object_name, "size": obj.size, "is_dir": obj.object_name.endswith("/"), "modifiedAt": obj.last_modified.isoformat() if obj.last_modified else None, }) return result def get_object(self, key: str) -> bytes: try: response = self._client.get_object(self._bucket, key) except S3Error as exc: if exc.code == "NoSuchKey": raise FileNotFoundError(key) from exc raise try: return response.read() finally: response.close() response.release_conn() def put_object(self, key: str, data: bytes, content_type: str) -> None: self._client.put_object( self._bucket, key, io.BytesIO(data), length=len(data), content_type=content_type, ) def delete_object(self, key: str) -> None: self._client.remove_object(self._bucket, key)