61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
"""Small S3-compatible object-storage adapter shared by backend features."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import boto3
|
|
from botocore.config import Config
|
|
from botocore.exceptions import ClientError
|
|
|
|
import settings
|
|
|
|
|
|
def _client():
|
|
return boto3.client(
|
|
"s3",
|
|
endpoint_url=settings.S3_ENDPOINT_URL,
|
|
aws_access_key_id=settings.S3_ACCESS_KEY,
|
|
aws_secret_access_key=settings.S3_SECRET_KEY,
|
|
region_name=settings.S3_REGION,
|
|
config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
|
|
)
|
|
|
|
|
|
def put_object(key: str, data: bytes, mime_type: str) -> None:
|
|
client = _client()
|
|
try:
|
|
client.head_bucket(Bucket=settings.S3_BUCKET)
|
|
except ClientError as exc:
|
|
status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
|
|
if status != 404:
|
|
raise
|
|
client.create_bucket(Bucket=settings.S3_BUCKET)
|
|
client.put_object(
|
|
Bucket=settings.S3_BUCKET,
|
|
Key=key,
|
|
Body=data,
|
|
ContentType=mime_type,
|
|
)
|
|
|
|
|
|
def get_object(key: str) -> bytes:
|
|
response = _client().get_object(Bucket=settings.S3_BUCKET, Key=key)
|
|
return response["Body"].read()
|
|
|
|
|
|
def delete_object(key: str) -> None:
|
|
_client().delete_object(Bucket=settings.S3_BUCKET, Key=key)
|
|
|
|
|
|
def storage_uri(key: str) -> str:
|
|
return f"s3://{settings.S3_BUCKET}/{key.lstrip('/')}"
|
|
|
|
|
|
def key_from_storage_uri(uri: str) -> str:
|
|
prefix = f"s3://{settings.S3_BUCKET}/"
|
|
if not uri.startswith(prefix):
|
|
raise ValueError("不支持的会话附件存储地址")
|
|
key = uri[len(prefix) :].strip("/")
|
|
if not key:
|
|
raise ValueError("会话附件存储地址缺少对象键")
|
|
return key
|