71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
|
import boto3
|
||
|
|
from botocore.exceptions import ClientError
|
||
|
|
|
||
|
|
|
||
|
|
class MinioClient:
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
endpoint_url: str,
|
||
|
|
access_key: str,
|
||
|
|
secret_key: str,
|
||
|
|
bucket: str,
|
||
|
|
api_base_url: str,
|
||
|
|
) -> None:
|
||
|
|
self._endpoint_url = endpoint_url
|
||
|
|
self._access_key = access_key
|
||
|
|
self._secret_key = secret_key
|
||
|
|
self._bucket = bucket
|
||
|
|
self._api_base_url = api_base_url.rstrip("/")
|
||
|
|
|
||
|
|
def _s3(self):
|
||
|
|
return boto3.client(
|
||
|
|
"s3",
|
||
|
|
endpoint_url=self._endpoint_url,
|
||
|
|
aws_access_key_id=self._access_key,
|
||
|
|
aws_secret_access_key=self._secret_key,
|
||
|
|
)
|
||
|
|
|
||
|
|
def ensure_bucket_exists(self) -> None:
|
||
|
|
client = self._s3()
|
||
|
|
try:
|
||
|
|
client.head_bucket(Bucket=self._bucket)
|
||
|
|
except ClientError as e:
|
||
|
|
if e.response["Error"]["Code"] in ("404", "NoSuchBucket"):
|
||
|
|
client.create_bucket(Bucket=self._bucket)
|
||
|
|
else:
|
||
|
|
raise
|
||
|
|
|
||
|
|
def upload(self, path: str, data: bytes) -> bool:
|
||
|
|
try:
|
||
|
|
self._s3().put_object(
|
||
|
|
Bucket=self._bucket,
|
||
|
|
Key=path,
|
||
|
|
Body=data,
|
||
|
|
ContentType="audio/wav",
|
||
|
|
)
|
||
|
|
return True
|
||
|
|
except ClientError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
def get_url(self, path: str) -> str:
|
||
|
|
return f"{self._api_base_url}/media/{path}"
|
||
|
|
|
||
|
|
def get_public_url(self, path: str) -> str:
|
||
|
|
return f"{self._api_base_url}/media/{path}"
|
||
|
|
|
||
|
|
def delete(self, path: str) -> bool:
|
||
|
|
try:
|
||
|
|
self._s3().delete_object(Bucket=self._bucket, Key=path)
|
||
|
|
return True
|
||
|
|
except ClientError:
|
||
|
|
return False
|
||
|
|
|
||
|
|
def download(self, path: str) -> tuple[bytes, str]:
|
||
|
|
try:
|
||
|
|
response = self._s3().get_object(Bucket=self._bucket, Key=path)
|
||
|
|
return response["Body"].read(), response.get("ContentType", "audio/wav")
|
||
|
|
except ClientError as e:
|
||
|
|
if e.response["Error"]["Code"] in ("NoSuchKey", "404"):
|
||
|
|
raise FileNotFoundError(path)
|
||
|
|
raise
|