admin
d703ab64e3
CI / docs-check (push) Canceled after 0s
CI / backend-test (push) Canceled after 0s
CI / service-test (push) Canceled after 0s
CI / frontend-test (push) Canceled after 0s
CI / rust-core (push) Canceled after 0s
CI / docs-check (pull_request) Canceled after 0s
CI / backend-test (pull_request) Canceled after 0s
CI / service-test (pull_request) Canceled after 0s
CI / frontend-test (pull_request) Canceled after 0s
CI / rust-core (pull_request) Canceled after 0s
128 lines
4.6 KiB
Python
128 lines
4.6 KiB
Python
"""内容对象按 Vault 分区；测试磁盘适配器不作为生产对象存储。"""
|
|
|
|
from pathlib import Path
|
|
import hashlib
|
|
import os
|
|
import tempfile
|
|
import shutil
|
|
from contextlib import contextmanager
|
|
|
|
|
|
class DiskObjects:
|
|
def __init__(self, root: Path):
|
|
self.root = root
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
|
|
def put(self, key: str, data: bytes):
|
|
target = self.root / key
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, temp = tempfile.mkstemp(dir=target.parent)
|
|
try:
|
|
with os.fdopen(fd, "wb") as stream:
|
|
stream.write(data)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temp, target)
|
|
finally:
|
|
Path(temp).unlink(missing_ok=True)
|
|
|
|
def get(self, key: str) -> bytes:
|
|
return (self.root / key).read_bytes()
|
|
|
|
def put_file(self, key: str, path: Path, content_hash: str):
|
|
target = self.root / key
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile(dir=target.parent, delete=False) as stream:
|
|
temporary = Path(stream.name)
|
|
try:
|
|
with path.open("rb") as source:
|
|
shutil.copyfileobj(source, stream, 1024 * 1024)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
except BaseException:
|
|
stream.close()
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|
|
try:
|
|
os.replace(temporary, target)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
@contextmanager
|
|
def open(self, key: str):
|
|
with (self.root / key).open("rb") as stream:
|
|
yield stream
|
|
|
|
def delete(self, key: str):
|
|
(self.root / key).unlink(missing_ok=True)
|
|
|
|
|
|
class S3Objects:
|
|
def __init__(self, endpoint: str, bucket: str):
|
|
import boto3
|
|
from botocore.config import Config
|
|
self.client = boto3.client("s3", endpoint_url=endpoint,
|
|
config=Config(connect_timeout=2, read_timeout=2,
|
|
retries={"max_attempts": 0}))
|
|
self.bucket = bucket
|
|
|
|
def ensure_bucket(self) -> bool:
|
|
"""不在时创建配置的桶；切勿更改现有存储桶。"""
|
|
from botocore.exceptions import ClientError
|
|
|
|
try:
|
|
self.client.head_bucket(Bucket=self.bucket)
|
|
return False
|
|
except ClientError as error:
|
|
code = str(error.response.get("Error", {}).get("Code", ""))
|
|
status = error.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
|
|
if code not in {"404", "NoSuchBucket", "NotFound"} and status != 404:
|
|
raise
|
|
self.client.create_bucket(Bucket=self.bucket)
|
|
return True
|
|
|
|
def is_empty(self) -> bool:
|
|
response = self.client.list_objects_v2(Bucket=self.bucket, MaxKeys=1)
|
|
return not response.get("Contents")
|
|
|
|
def delete_many(self, keys: list[str]) -> None:
|
|
for start in range(0, len(keys), 1000):
|
|
batch = keys[start : start + 1000]
|
|
if batch:
|
|
self.client.delete_objects(
|
|
Bucket=self.bucket,
|
|
Delete={"Objects": [{"Key": key} for key in batch], "Quiet": True},
|
|
)
|
|
|
|
def delete_bucket(self) -> None:
|
|
self.client.delete_bucket(Bucket=self.bucket)
|
|
|
|
def put(self, key: str, data: bytes):
|
|
self.client.put_object(Bucket=self.bucket, Key=key, Body=data,
|
|
Metadata={"sha256": hashlib.sha256(data).hexdigest()})
|
|
|
|
def get(self, key: str) -> bytes:
|
|
response = self.client.get_object(Bucket=self.bucket, Key=key)
|
|
with response["Body"] as stream:
|
|
return stream.read()
|
|
|
|
def put_file(self, key: str, path: Path, content_hash: str):
|
|
with path.open("rb") as stream:
|
|
# 对象的上限为 100 MiB，远低于 S3 的 5 GiB 单 PUT 限制。直接流请求有一个显式的连接生命周期；每次完成构建一个传输管理器可以在重复的多工作线程使用下保留池化的 MinIO 连接。
|
|
self.client.put_object(
|
|
Bucket=self.bucket,
|
|
Key=key,
|
|
Body=stream,
|
|
ContentLength=path.stat().st_size,
|
|
Metadata={"sha256": content_hash},
|
|
)
|
|
|
|
@contextmanager
|
|
def open(self, key: str):
|
|
response = self.client.get_object(Bucket=self.bucket, Key=key)
|
|
with response["Body"] as stream:
|
|
yield stream
|
|
|
|
def delete(self, key: str):
|
|
self.client.delete_object(Bucket=self.bucket, Key=key)
|