refactor(common.storage): drop sync storage abstraction and example_usage
The async-only direction was already the only one used in production: * create_storage never accepted mode=sync; build_storage_config always emitted mode=async; zero callers referenced StorageBackend / SyncData / S3StorageBackend.sync / LocalStorageBackend.sync anywhere. * Drop the parallel sync base class, the sync concrete classes in backends/local.py and backends/s3.py, and the boto3 dependency. * Drop example_usage.py (zero importers; demonstration code, not part of the public surface). * Rename LocalAsyncStorageBackend -> LocalStorageBackend, S3AsyncStorageBackend -> S3StorageBackend to reflect the single remaining class per type. * Tighten create_storage: any mode=... key now raises StorageConfigError with the new pointer (settings.storage_backend controls behavior). * Cleanup call sites: schedule.application.service.build_object_store no longer passes mode=async to create_storage. * Cosmetic touch-ups in backend/services/storage.py and common/config.py docstrings where they still said "boto3" instead of "S3 client". Public API surface preserved: AsyncStorageBackend / ObjectMeta / create_storage / build_storage_config / register_backend all keep their names and call signatures. backend tests: 136 passed.
This commit is contained in:
tao.chen
@@ -680,7 +680,7 @@ async def create_download_url_payload(
|
||||
expires_in=timedelta(seconds=payload.expires_seconds),
|
||||
)
|
||||
# Public-host rewriting is now nginx's job (location /storage/). In the
|
||||
# future the boto3 client should be built with the public endpoint so
|
||||
# future the S3 client should be built with the public endpoint so
|
||||
# generate_presigned_url returns a public URL directly.
|
||||
return {
|
||||
"data": {
|
||||
|
||||
@@ -7,7 +7,6 @@ dependencies = [
|
||||
"greenlet>=3.0.0",
|
||||
"apscheduler>=3.11.3",
|
||||
"asyncmy==0.2.11",
|
||||
"boto3>=1.34,<2",
|
||||
"fastapi==0.116.1",
|
||||
"pydantic-settings>=2.14.2",
|
||||
"loguru>=0.7.2",
|
||||
|
||||
@@ -139,11 +139,11 @@ class Settings(BaseSettings):
|
||||
)
|
||||
s3_access_key: str = Field(
|
||||
default="modelplatform",
|
||||
description="boto3 access key for S3-compatible storage.",
|
||||
description="S3 access key for S3-compatible storage.",
|
||||
)
|
||||
s3_secret_key: str = Field(
|
||||
default="modelplatformsecret",
|
||||
description="boto3 secret key for S3-compatible storage.",
|
||||
description="S3 secret key for S3-compatible storage.",
|
||||
)
|
||||
s3_workspace_bucket: str = Field(
|
||||
default="workspace",
|
||||
|
||||
@@ -1,24 +1,17 @@
|
||||
"""统一存储层，同时支持同步和异步，通过 config["mode"] 切换。
|
||||
"""异步统一存储层。
|
||||
|
||||
对上层暴露的公开 API：
|
||||
对外暴露的公开 API：
|
||||
|
||||
from storage import create_storage, StorageBackend, AsyncStorageBackend, ObjectMeta
|
||||
from storage.exceptions import StorageError, StorageNotFoundError, ...
|
||||
from common.storage import create_storage, AsyncStorageBackend, ObjectMeta
|
||||
from common.storage.exceptions import StorageError, StorageNotFoundError, ...
|
||||
|
||||
切换本地 / S3 通过 ``settings.storage_backend`` 控制，业务代码不感知差异：
|
||||
|
||||
用法：
|
||||
# 同步（默认 mode="sync"）
|
||||
storage = create_storage({"type": "local", "base_dir": "./data"})
|
||||
storage.put("a/b.txt", b"hello")
|
||||
|
||||
# 异步：加一个 mode 字段
|
||||
storage = create_storage({"type": "local", "mode": "async", "base_dir": "./data"})
|
||||
await storage.put("a/b.txt", b"hello")
|
||||
|
||||
切换本地/S3，或切换同步/异步，业务代码都不用改，只改配置：
|
||||
storage = create_storage({"type": "s3", "mode": "async", "bucket": "my-bucket"})
|
||||
storage = create_storage({"type": "s3", "bucket": "my-bucket"})
|
||||
"""
|
||||
|
||||
from .base import AsyncStorageBackend, ObjectMeta, StorageBackend
|
||||
from .base import AsyncStorageBackend, ObjectMeta
|
||||
from .factory import (
|
||||
PURPOSE_BUCKETS,
|
||||
RCLONE_REMOTE_NAME,
|
||||
@@ -38,7 +31,6 @@ __all__ = [
|
||||
"USAGE_TYPE_TO_PURPOSE",
|
||||
"AsyncStorageBackend",
|
||||
"ObjectMeta",
|
||||
"StorageBackend",
|
||||
"actual_bucket_name",
|
||||
"build_storage_config",
|
||||
"build_storage_uri",
|
||||
|
||||
@@ -1,21 +1,20 @@
|
||||
"""本地文件系统存储后端。
|
||||
"""本地文件系统异步存储后端。
|
||||
|
||||
- 同步实现 `LocalStorageBackend`：标准库文件 I/O
|
||||
- 异步实现 `LocalAsyncStorageBackend`：aiofiles 做实际读写，
|
||||
stat/exists/delete/mkdir/目录遍历这类轻量元数据操作用
|
||||
asyncio.to_thread 包一层，避免阻塞事件循环
|
||||
（只有创建异步实例时才需要装 aiofiles，同步实现零依赖）
|
||||
`LocalStorageBackend`：使用 aiofiles 做实际读写，
|
||||
stat/exists/delete/mkdir/目录遍历这类轻量元数据操作用
|
||||
``asyncio.to_thread`` 包一层，避免阻塞事件循环。
|
||||
|
||||
依赖：pip install aiofiles
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import shutil
|
||||
from collections.abc import AsyncIterator, Iterable
|
||||
from collections.abc import AsyncIterator
|
||||
from datetime import timedelta
|
||||
from pathlib import Path
|
||||
from typing import BinaryIO
|
||||
|
||||
from ..base import AsyncData, AsyncStorageBackend, ObjectMeta, StorageBackend, SyncData
|
||||
from ..base import AsyncData, AsyncStorageBackend, ObjectMeta
|
||||
from ..exceptions import StorageAlreadyExistsError, StorageNotFoundError
|
||||
from ..registry import register_backend
|
||||
|
||||
@@ -33,106 +32,9 @@ def _meta(key: str, path: Path) -> ObjectMeta:
|
||||
return ObjectMeta(key=key, size=st.st_size, last_modified=st.st_mtime)
|
||||
|
||||
|
||||
# ==================== 同步实现 ====================
|
||||
|
||||
|
||||
@register_backend("local", mode="sync")
|
||||
class LocalStorageBackend(StorageBackend):
|
||||
"""配置示例: {"type": "local", "mode": "sync", "base_dir": "/data/storage"}"""
|
||||
|
||||
def __init__(self, base_dir: str, **_ignored):
|
||||
self.base_dir = Path(base_dir).resolve()
|
||||
self.base_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _resolve(self, key: str) -> Path:
|
||||
return _resolve(self.base_dir, key)
|
||||
|
||||
def put(
|
||||
self,
|
||||
key: str,
|
||||
data: SyncData,
|
||||
*,
|
||||
overwrite: bool = True,
|
||||
content_type: str | None = None,
|
||||
metadata: dict | None = None,
|
||||
) -> ObjectMeta:
|
||||
path = self._resolve(key)
|
||||
if path.exists() and not overwrite:
|
||||
raise StorageAlreadyExistsError(f"key 已存在: {key}")
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if isinstance(data, bytes):
|
||||
path.write_bytes(data)
|
||||
else:
|
||||
with open(path, "wb") as f:
|
||||
shutil.copyfileobj(data, f)
|
||||
# local FS 没有对象级 metadata；content_type / metadata 暂存忽略。
|
||||
return _meta(key, path)
|
||||
|
||||
def get(self, key: str) -> bytes:
|
||||
path = self._resolve(key)
|
||||
if not path.is_file():
|
||||
raise StorageNotFoundError(f"key 不存在: {key}")
|
||||
return path.read_bytes()
|
||||
|
||||
def get_stream(self, key: str) -> BinaryIO:
|
||||
path = self._resolve(key)
|
||||
if not path.is_file():
|
||||
raise StorageNotFoundError(f"key 不存在: {key}")
|
||||
return open(path, "rb")
|
||||
|
||||
def delete(self, key: str) -> None:
|
||||
try:
|
||||
self._resolve(key).unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
def exists(self, key: str) -> bool:
|
||||
return self._resolve(key).is_file()
|
||||
|
||||
def stat(self, key: str) -> ObjectMeta:
|
||||
path = self._resolve(key)
|
||||
if not path.is_file():
|
||||
raise StorageNotFoundError(f"key 不存在: {key}")
|
||||
return _meta(key, path)
|
||||
|
||||
def list(self, prefix: str = "") -> Iterable[ObjectMeta]:
|
||||
search_root = self._resolve(prefix) if prefix else self.base_dir
|
||||
if search_root.is_dir():
|
||||
candidates = search_root.rglob("*")
|
||||
else:
|
||||
candidates = search_root.parent.glob(f"{search_root.name}*")
|
||||
|
||||
for path in candidates:
|
||||
if path.is_file():
|
||||
key = str(path.relative_to(self.base_dir)).replace(os.sep, "/")
|
||||
yield _meta(key, path)
|
||||
|
||||
def get_url(self, key: str, *, expires_in: timedelta | None = None) -> str:
|
||||
path = self._resolve(key)
|
||||
if not path.is_file():
|
||||
raise StorageNotFoundError(f"key 不存在: {key}")
|
||||
return path.as_uri()
|
||||
|
||||
def copy(self, src_key: str, dst_key: str) -> ObjectMeta:
|
||||
src_path = self._resolve(src_key)
|
||||
if not src_path.is_file():
|
||||
raise StorageNotFoundError(f"key 不存在: {src_key}")
|
||||
dst_path = self._resolve(dst_key)
|
||||
dst_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(src_path, dst_path)
|
||||
return _meta(dst_key, dst_path)
|
||||
|
||||
|
||||
# ==================== 异步实现 ====================
|
||||
|
||||
|
||||
@register_backend("local", mode="async")
|
||||
class LocalAsyncStorageBackend(AsyncStorageBackend):
|
||||
"""配置示例: {"type": "local", "mode": "async", "base_dir": "/data/storage"}
|
||||
|
||||
需要: pip install aiofiles
|
||||
"""
|
||||
@register_backend("local")
|
||||
class LocalStorageBackend(AsyncStorageBackend):
|
||||
"""配置示例: {"type": "local", "base_dir": "/data/storage"}"""
|
||||
|
||||
def __init__(self, base_dir: str, **_ignored):
|
||||
self.base_dir = Path(base_dir).resolve()
|
||||
|
||||
@@ -1,18 +1,15 @@
|
||||
"""S3（及兼容协议）存储后端。
|
||||
"""S3（及兼容协议）异步存储后端。
|
||||
|
||||
- 同步实现 `S3StorageBackend`：boto3
|
||||
- 异步实现 `S3AsyncStorageBackend`：aioboto3
|
||||
`S3StorageBackend`：用 aioboto3 跑所有 S3 操作；
|
||||
异常类型从 ``botocore.exceptions`` 拿（aioboto3 透传）。
|
||||
|
||||
两者只在各自 __init__ 里做 lazy import，互不强制依赖：
|
||||
只用同步模式不需要装 aioboto3，只用异步模式不需要额外装 boto3
|
||||
（aioboto3 本身依赖 botocore，异常类型从它里面拿）。
|
||||
依赖：pip install aioboto3
|
||||
"""
|
||||
|
||||
from collections.abc import AsyncIterator, Iterable
|
||||
from collections.abc import AsyncIterator
|
||||
from datetime import timedelta
|
||||
from typing import BinaryIO
|
||||
|
||||
from ..base import AsyncData, AsyncStorageBackend, ObjectMeta, StorageBackend, SyncData
|
||||
from ..base import AsyncData, AsyncStorageBackend, ObjectMeta
|
||||
from ..exceptions import (
|
||||
StorageAlreadyExistsError,
|
||||
StorageConnectionError,
|
||||
@@ -30,162 +27,16 @@ def _meta_from_head(key: str, head: dict) -> ObjectMeta:
|
||||
)
|
||||
|
||||
|
||||
# ==================== 同步实现 ====================
|
||||
|
||||
|
||||
@register_backend("s3", mode="sync")
|
||||
class S3StorageBackend(StorageBackend):
|
||||
@register_backend("s3")
|
||||
class S3StorageBackend(AsyncStorageBackend):
|
||||
"""配置示例:
|
||||
{
|
||||
"type": "s3", "mode": "sync",
|
||||
"type": "s3",
|
||||
"bucket": "my-bucket", "prefix": "app1/",
|
||||
"region_name": "cn-north-1", "endpoint_url": "https://s3.example.com",
|
||||
"aws_access_key_id": "...", "aws_secret_access_key": "...",
|
||||
}
|
||||
|
||||
需要: pip install boto3
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
bucket: str,
|
||||
prefix: str = "",
|
||||
region_name: str | None = None,
|
||||
endpoint_url: str | None = None,
|
||||
aws_access_key_id: str | None = None,
|
||||
aws_secret_access_key: str | None = None,
|
||||
**_ignored,
|
||||
):
|
||||
try:
|
||||
import boto3
|
||||
from botocore.exceptions import BotoCoreError, ClientError
|
||||
except ImportError as e:
|
||||
raise ImportError("使用同步 S3 存储后端需要先安装 boto3: pip install boto3") from e
|
||||
|
||||
self._ClientError = ClientError
|
||||
self._BotoCoreError = BotoCoreError
|
||||
self.bucket = bucket
|
||||
self.prefix = prefix.strip("/") + "/" if prefix.strip("/") else ""
|
||||
|
||||
try:
|
||||
self.client = boto3.client(
|
||||
"s3",
|
||||
region_name=region_name,
|
||||
endpoint_url=endpoint_url,
|
||||
aws_access_key_id=aws_access_key_id,
|
||||
aws_secret_access_key=aws_secret_access_key,
|
||||
)
|
||||
except (BotoCoreError, ClientError) as e:
|
||||
raise StorageConnectionError(f"初始化 S3 client 失败: {e}") from e
|
||||
|
||||
def _full_key(self, key: str) -> str:
|
||||
return f"{self.prefix}{key.lstrip('/')}"
|
||||
|
||||
def put(self, key: str, data: SyncData, *, overwrite: bool = True) -> ObjectMeta:
|
||||
full_key = self._full_key(key)
|
||||
if not overwrite and self.exists(key):
|
||||
raise StorageAlreadyExistsError(f"key 已存在: {key}")
|
||||
body = data if isinstance(data, bytes) else data.read()
|
||||
try:
|
||||
self.client.put_object(Bucket=self.bucket, Key=full_key, Body=body)
|
||||
except (self._ClientError, self._BotoCoreError) as e:
|
||||
raise StorageConnectionError(f"上传失败 key={key}: {e}") from e
|
||||
return self.stat(key)
|
||||
|
||||
def get(self, key: str) -> bytes:
|
||||
full_key = self._full_key(key)
|
||||
try:
|
||||
resp = self.client.get_object(Bucket=self.bucket, Key=full_key)
|
||||
return resp["Body"].read()
|
||||
except self._ClientError as e:
|
||||
if e.response.get("Error", {}).get("Code") in ("NoSuchKey", "404"):
|
||||
raise StorageNotFoundError(f"key 不存在: {key}") from e
|
||||
raise StorageConnectionError(f"读取失败 key={key}: {e}") from e
|
||||
|
||||
def get_stream(self, key: str) -> BinaryIO:
|
||||
full_key = self._full_key(key)
|
||||
try:
|
||||
resp = self.client.get_object(Bucket=self.bucket, Key=full_key)
|
||||
return resp["Body"]
|
||||
except self._ClientError as e:
|
||||
if e.response.get("Error", {}).get("Code") in ("NoSuchKey", "404"):
|
||||
raise StorageNotFoundError(f"key 不存在: {key}") from e
|
||||
raise StorageConnectionError(f"读取失败 key={key}: {e}") from e
|
||||
|
||||
def delete(self, key: str) -> None:
|
||||
try:
|
||||
self.client.delete_object(Bucket=self.bucket, Key=self._full_key(key))
|
||||
except (self._ClientError, self._BotoCoreError) as e:
|
||||
raise StorageConnectionError(f"删除失败 key={key}: {e}") from e
|
||||
|
||||
def exists(self, key: str) -> bool:
|
||||
try:
|
||||
self.client.head_object(Bucket=self.bucket, Key=self._full_key(key))
|
||||
return True
|
||||
except self._ClientError as e:
|
||||
if e.response.get("Error", {}).get("Code") in ("404", "NoSuchKey"):
|
||||
return False
|
||||
raise StorageConnectionError(f"检查 exists 失败 key={key}: {e}") from e
|
||||
|
||||
def stat(self, key: str) -> ObjectMeta:
|
||||
try:
|
||||
head = self.client.head_object(Bucket=self.bucket, Key=self._full_key(key))
|
||||
except self._ClientError as e:
|
||||
if e.response.get("Error", {}).get("Code") in ("404", "NoSuchKey"):
|
||||
raise StorageNotFoundError(f"key 不存在: {key}") from e
|
||||
raise StorageConnectionError(f"获取元信息失败 key={key}: {e}") from e
|
||||
return _meta_from_head(key, head)
|
||||
|
||||
def list(self, prefix: str = "") -> Iterable[ObjectMeta]:
|
||||
full_prefix = self._full_key(prefix)
|
||||
paginator = self.client.get_paginator("list_objects_v2")
|
||||
try:
|
||||
for page in paginator.paginate(Bucket=self.bucket, Prefix=full_prefix):
|
||||
for obj in page.get("Contents", []):
|
||||
key = obj["Key"][len(self.prefix):] if self.prefix else obj["Key"]
|
||||
yield ObjectMeta(
|
||||
key=key,
|
||||
size=obj["Size"],
|
||||
last_modified=obj["LastModified"].timestamp(),
|
||||
etag=obj.get("ETag"),
|
||||
)
|
||||
except (self._ClientError, self._BotoCoreError) as e:
|
||||
raise StorageConnectionError(f"列举对象失败 prefix={prefix}: {e}") from e
|
||||
|
||||
def get_url(self, key: str, *, expires_in: timedelta | None = None) -> str:
|
||||
expires_seconds = int(expires_in.total_seconds()) if expires_in else 3600
|
||||
try:
|
||||
return self.client.generate_presigned_url(
|
||||
"get_object",
|
||||
Params={"Bucket": self.bucket, "Key": self._full_key(key)},
|
||||
ExpiresIn=expires_seconds,
|
||||
)
|
||||
except (self._ClientError, self._BotoCoreError) as e:
|
||||
raise StorageConnectionError(f"生成预签名 URL 失败 key={key}: {e}") from e
|
||||
|
||||
def copy(self, src_key: str, dst_key: str) -> ObjectMeta:
|
||||
try:
|
||||
self.client.copy_object(
|
||||
Bucket=self.bucket,
|
||||
Key=self._full_key(dst_key),
|
||||
CopySource={"Bucket": self.bucket, "Key": self._full_key(src_key)},
|
||||
)
|
||||
except self._ClientError as e:
|
||||
if e.response.get("Error", {}).get("Code") in ("404", "NoSuchKey"):
|
||||
raise StorageNotFoundError(f"key 不存在: {src_key}") from e
|
||||
raise StorageConnectionError(f"复制失败 {src_key} -> {dst_key}: {e}") from e
|
||||
return self.stat(dst_key)
|
||||
|
||||
|
||||
# ==================== 异步实现 ====================
|
||||
|
||||
|
||||
@register_backend("s3", mode="async")
|
||||
class S3AsyncStorageBackend(AsyncStorageBackend):
|
||||
"""配置示例同上，把 "mode" 改成 "async" 即可。
|
||||
|
||||
需要: pip install aioboto3
|
||||
|
||||
每次操作默认通过 `async with session.client(...)` 拿一个短生命周期
|
||||
client；用 `async with create_storage(...) as storage:` 可以复用同一个
|
||||
client（见 __aenter__/__aexit__）。
|
||||
@@ -226,7 +77,7 @@ class S3AsyncStorageBackend(AsyncStorageBackend):
|
||||
def _client_cm(self):
|
||||
return self._session.client("s3", **self._client_kwargs)
|
||||
|
||||
async def __aenter__(self) -> "S3AsyncStorageBackend":
|
||||
async def __aenter__(self) -> "S3StorageBackend":
|
||||
self._persistent_cm = self._client_cm()
|
||||
self._persistent_client = await self._persistent_cm.__aenter__()
|
||||
return self
|
||||
|
||||
@@ -1,21 +1,18 @@
|
||||
"""同步 / 异步存储后端统一抽象接口。
|
||||
"""统一异步存储后端抽象接口。
|
||||
|
||||
`StorageBackend` 是同步接口，`AsyncStorageBackend` 是异步接口，
|
||||
两者共用同一个 `ObjectMeta` 数据结构，方法签名尽量保持对称
|
||||
（异步版本每个方法多一个 await，get_stream/list 变成异步生成器），
|
||||
这样业务代码从同步切到异步时心智负担最小。
|
||||
所有异步存储后端(local / s3 / 未来新加的)继承 ``AsyncStorageBackend``，
|
||||
方法签名共用同一个 ``ObjectMeta`` 返回结构。
|
||||
|
||||
上层通过 `storage.create_storage(config)` 统一创建实例，
|
||||
用 `config["mode"]` 决定拿到的是同步实现还是异步实现。
|
||||
上层通过 ``storage.create_storage(config)`` 统一创建实例；
|
||||
不再提供同步抽象 —— 所有调用方都使用 ``async`` 接口。
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from collections.abc import AsyncIterator, Iterable
|
||||
from collections.abc import AsyncIterator
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import timedelta
|
||||
from typing import BinaryIO, Union
|
||||
from typing import Union
|
||||
|
||||
SyncData = Union[bytes, BinaryIO]
|
||||
AsyncData = Union[bytes, "AsyncIterator[bytes]"]
|
||||
|
||||
|
||||
@@ -30,68 +27,6 @@ class ObjectMeta:
|
||||
extra: dict = field(default_factory=dict) # 后端特有的额外信息
|
||||
|
||||
|
||||
class StorageBackend(ABC):
|
||||
"""同步存储后端统一抽象基类。"""
|
||||
|
||||
@abstractmethod
|
||||
def put(
|
||||
self,
|
||||
key: str,
|
||||
data: SyncData,
|
||||
*,
|
||||
overwrite: bool = True,
|
||||
content_type: str | None = None,
|
||||
metadata: dict | None = None,
|
||||
) -> ObjectMeta:
|
||||
"""写入对象。overwrite=False 时 key 已存在应抛出 StorageAlreadyExistsError。
|
||||
|
||||
``content_type`` 和 ``metadata`` 是可选的（与异步 put 语义一致）。
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def get(self, key: str) -> bytes:
|
||||
"""读取对象内容，不存在时抛出 StorageNotFoundError。"""
|
||||
|
||||
@abstractmethod
|
||||
def get_stream(self, key: str) -> BinaryIO:
|
||||
"""以流方式读取对象，适合大文件。"""
|
||||
|
||||
@abstractmethod
|
||||
def delete(self, key: str) -> None:
|
||||
"""删除对象。删除不存在的 key 不应报错（幂等）。"""
|
||||
|
||||
@abstractmethod
|
||||
def exists(self, key: str) -> bool:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def stat(self, key: str) -> ObjectMeta:
|
||||
"""不存在时抛出 StorageNotFoundError。"""
|
||||
|
||||
@abstractmethod
|
||||
def list(self, prefix: str = "") -> Iterable[ObjectMeta]:
|
||||
"""按前缀列出对象。"""
|
||||
|
||||
@abstractmethod
|
||||
def get_url(self, key: str, *, expires_in: timedelta | None = None) -> str:
|
||||
"""获取可访问 URL；本地存储返回 file://，S3 返回预签名 URL。"""
|
||||
|
||||
def copy(self, src_key: str, dst_key: str) -> ObjectMeta:
|
||||
"""默认实现：读出来再写进去。后端可覆盖为更高效的原生实现。"""
|
||||
data = self.get(src_key)
|
||||
return self.put(dst_key, data)
|
||||
|
||||
def close(self) -> None:
|
||||
"""释放后端持有的资源（连接池等）。不需要的后端可以不覆盖。"""
|
||||
return
|
||||
|
||||
def __enter__(self) -> "StorageBackend":
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb) -> None:
|
||||
self.close()
|
||||
|
||||
|
||||
class AsyncStorageBackend(ABC):
|
||||
"""异步存储后端统一抽象基类。"""
|
||||
|
||||
|
||||
@@ -1,115 +0,0 @@
|
||||
"""使用示例：同一套 create_storage()，靠 config["mode"] 切换同步/异步。"""
|
||||
|
||||
import asyncio
|
||||
|
||||
from common.storage import create_storage
|
||||
from common.storage.exceptions import StorageNotFoundError
|
||||
|
||||
|
||||
def sync_demo():
|
||||
# mode 默认就是 "sync"，可以不写
|
||||
storage = create_storage({"type": "local", "base_dir": "./data_sync"})
|
||||
|
||||
storage.put("docs/hello.txt", b"hello world")
|
||||
print(storage.get("docs/hello.txt"))
|
||||
print(storage.exists("docs/hello.txt"))
|
||||
print(list(storage.list("docs/")))
|
||||
print(storage.get_url("docs/hello.txt"))
|
||||
|
||||
try:
|
||||
storage.get("docs/not_exist.txt")
|
||||
except StorageNotFoundError:
|
||||
print("按预期抛出 StorageNotFoundError")
|
||||
|
||||
# 换成同步 S3，只改配置：
|
||||
# storage = create_storage({"type": "s3", "bucket": "my-bucket"})
|
||||
|
||||
|
||||
async def async_demo():
|
||||
# 只加一个 "mode": "async"，其余配置和参数不变
|
||||
storage = create_storage({"type": "local", "mode": "async", "base_dir": "./data_async"})
|
||||
|
||||
await storage.put("docs/hello.txt", b"hello world")
|
||||
print(await storage.get("docs/hello.txt"))
|
||||
print(await storage.exists("docs/hello.txt"))
|
||||
|
||||
async for meta in storage.list("docs/"):
|
||||
print(meta)
|
||||
|
||||
chunks = []
|
||||
async for chunk in storage.get_stream("docs/hello.txt"):
|
||||
chunks.append(chunk)
|
||||
print(b"".join(chunks))
|
||||
|
||||
try:
|
||||
await storage.get("docs/not_exist.txt")
|
||||
except StorageNotFoundError:
|
||||
print("按预期抛出 StorageNotFoundError")
|
||||
|
||||
# 并发写入，异步模式的典型优势场景
|
||||
tasks = [storage.put(f"batch/{i}.txt", f"content-{i}".encode()) for i in range(10)]
|
||||
await asyncio.gather(*tasks)
|
||||
print("并发写入 10 个对象完成")
|
||||
|
||||
# 换成异步 S3，只改配置：
|
||||
# storage = create_storage({"type": "s3", "mode": "async", "bucket": "my-bucket"})
|
||||
# 高吞吐场景复用连接：
|
||||
# async with create_storage({"type": "s3", "mode": "async", "bucket": "my-bucket"}) as s3:
|
||||
# await s3.put("a.txt", b"1")
|
||||
|
||||
|
||||
def extend_with_new_backend_demo():
|
||||
"""演示独立扩展一种新的存储方式（同步+异步各一个），不用改现有代码。"""
|
||||
import io
|
||||
import time
|
||||
|
||||
from common.storage.base import ObjectMeta, StorageBackend
|
||||
from common.storage.exceptions import StorageNotFoundError
|
||||
from common.storage.registry import register_backend
|
||||
|
||||
@register_backend("memory", mode="sync")
|
||||
class MemoryStorageBackend(StorageBackend):
|
||||
def __init__(self, **_ignored):
|
||||
self._store = {}
|
||||
|
||||
def put(self, key, data, *, overwrite=True):
|
||||
body = data if isinstance(data, bytes) else data.read()
|
||||
self._store[key] = body
|
||||
return ObjectMeta(key=key, size=len(body), last_modified=time.time())
|
||||
|
||||
def get(self, key):
|
||||
if key not in self._store:
|
||||
raise StorageNotFoundError(key)
|
||||
return self._store[key]
|
||||
|
||||
def get_stream(self, key):
|
||||
return io.BytesIO(self.get(key))
|
||||
|
||||
def delete(self, key):
|
||||
self._store.pop(key, None)
|
||||
|
||||
def exists(self, key):
|
||||
return key in self._store
|
||||
|
||||
def stat(self, key):
|
||||
if key not in self._store:
|
||||
raise StorageNotFoundError(key)
|
||||
return ObjectMeta(key=key, size=len(self._store[key]))
|
||||
|
||||
def list(self, prefix=""):
|
||||
for key, body in self._store.items():
|
||||
if key.startswith(prefix):
|
||||
yield ObjectMeta(key=key, size=len(body))
|
||||
|
||||
def get_url(self, key, *, expires_in=None):
|
||||
return f"memory://{key}"
|
||||
|
||||
mem_storage = create_storage({"type": "memory", "mode": "sync"})
|
||||
mem_storage.put("a.txt", b"in-memory content")
|
||||
print(mem_storage.get("a.txt"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sync_demo()
|
||||
asyncio.run(async_demo())
|
||||
extend_with_new_backend_demo()
|
||||
@@ -1,58 +1,56 @@
|
||||
"""统一入口：根据配置字典创建具体的存储后端实例。
|
||||
"""统一入口：根据配置字典创建具体的异步存储后端实例。
|
||||
|
||||
配置里的 "mode" 字段决定拿到同步还是异步实现，默认 "sync"（向后兼容）。
|
||||
通过 ``settings.storage_backend`` 切换本地 / S3，业务代码不感知差异。
|
||||
|
||||
# 同步（默认）
|
||||
# 本地
|
||||
storage = create_storage({"type": "local", "base_dir": "./data"})
|
||||
storage.put("a.txt", b"hello")
|
||||
|
||||
# 异步：只需加一个 mode 字段，其余配置不变
|
||||
storage = create_storage({"type": "local", "mode": "async", "base_dir": "./data"})
|
||||
await storage.put("a.txt", b"hello")
|
||||
|
||||
# S3 同理
|
||||
storage = create_storage({"type": "s3", "mode": "async", "bucket": "my-bucket"})
|
||||
# S3
|
||||
storage = create_storage({"type": "s3", "bucket": "my-bucket"})
|
||||
await storage.put("a.txt", b"hello")
|
||||
|
||||
上层业务代码应该只从这里拿实例，不要直接 import 具体的 XxxStorageBackend /
|
||||
XxxAsyncStorageBackend 类。
|
||||
上层业务代码应该只从这里拿实例，不要直接 import 具体的 XxxStorageBackend 类。
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Union
|
||||
from typing import Any
|
||||
|
||||
from .backends import local, s3 # noqa: F401 # 触发内置后端注册
|
||||
from .base import AsyncStorageBackend, StorageBackend
|
||||
from .base import AsyncStorageBackend
|
||||
from .exceptions import StorageConfigError
|
||||
from .registry import get_backend_class
|
||||
|
||||
AnyStorageBackend = Union[StorageBackend, AsyncStorageBackend]
|
||||
|
||||
|
||||
def create_storage(config: dict[str, Any]) -> AnyStorageBackend:
|
||||
"""根据配置创建存储后端。
|
||||
def create_storage(config: dict[str, Any]) -> AsyncStorageBackend:
|
||||
"""根据配置创建异步存储后端。
|
||||
|
||||
Args:
|
||||
config: 必须包含 "type" 字段（如 "local" / "s3"）；
|
||||
可选 "mode" 字段（"sync" 默认 / "async"）；
|
||||
其余字段作为 kwargs 传给对应后端的构造函数。
|
||||
不能包含 "mode" ——同步接口已删除；如需切换行为请改 settings。
|
||||
|
||||
Returns:
|
||||
mode="sync" 时返回 StorageBackend 实例（同步方法）；
|
||||
mode="async" 时返回 AsyncStorageBackend 实例（方法需要 await）。
|
||||
``AsyncStorageBackend`` 实例（方法需要 await）。
|
||||
"""
|
||||
config = dict(config) # 不修改调用方传入的原字典
|
||||
backend_type = config.pop("type", None)
|
||||
mode = config.pop("mode", "sync")
|
||||
|
||||
if "mode" in config:
|
||||
raise StorageConfigError(
|
||||
"create_storage 不再接受 'mode' 字段；只支持异步后端。"
|
||||
"如需切换本地 / S3，请改 settings.storage_backend。"
|
||||
)
|
||||
|
||||
backend_type = config.pop("type", None)
|
||||
if not backend_type:
|
||||
raise StorageConfigError("配置缺少 'type' 字段，例如 'local' 或 's3'")
|
||||
|
||||
backend_cls = get_backend_class(backend_type, mode)
|
||||
backend_cls = get_backend_class(backend_type)
|
||||
try:
|
||||
return backend_cls(**config)
|
||||
except TypeError as e:
|
||||
raise StorageConfigError(
|
||||
f"创建后端 (mode={mode}, type={backend_type}) 失败，参数不匹配: {e}"
|
||||
f"创建后端 (type={backend_type}) 失败，参数不匹配: {e}"
|
||||
) from e
|
||||
|
||||
|
||||
@@ -125,7 +123,7 @@ def build_storage_config(bucket_name: str) -> dict[str, Any]:
|
||||
bucket_name: 桶名，必须是 ``PURPOSE_BUCKETS`` 之一。
|
||||
|
||||
Returns:
|
||||
直接喂给 ``create_storage(...)`` 的 dict。
|
||||
直接喂给 ``create_storage(...)`` 的 dict。不含 ``mode`` 字段。
|
||||
"""
|
||||
# 延迟 import：避免 storage -> config -> storage 的循环依赖
|
||||
from common.config import settings
|
||||
@@ -138,14 +136,12 @@ def build_storage_config(bucket_name: str) -> dict[str, Any]:
|
||||
if settings.storage_backend == "local":
|
||||
return {
|
||||
"type": "local",
|
||||
"mode": "async",
|
||||
"base_dir": str(Path(settings.local_storage_base_dir) / bucket_name),
|
||||
}
|
||||
|
||||
if settings.storage_backend == "s3":
|
||||
return {
|
||||
"type": "s3",
|
||||
"mode": "async",
|
||||
"bucket": getattr(settings, f"s3_{bucket_name}_bucket"),
|
||||
"endpoint_url": settings.s3_endpoint,
|
||||
"aws_access_key_id": settings.s3_access_key,
|
||||
|
||||
@@ -1,61 +1,43 @@
|
||||
"""后端注册表，用 (mode, name) 作为 key 同时管理同步和异步实现。
|
||||
"""后端注册表，按 name 索引每个后端的异步实现。
|
||||
|
||||
新增一种存储方式的同步或异步实现时，不需要改 factory.py：
|
||||
@register_backend("local", mode="sync")
|
||||
class LocalStorageBackend(StorageBackend): ...
|
||||
新增一种存储方式时，不需要改 factory.py：
|
||||
@register_backend("local")
|
||||
class LocalStorageBackend(AsyncStorageBackend): ...
|
||||
|
||||
@register_backend("local", mode="async")
|
||||
class LocalAsyncStorageBackend(AsyncStorageBackend): ...
|
||||
|
||||
只要保证模块被 import 一次即可（backends/__init__.py 里统一 import）。
|
||||
只要保证模块被 import 一次即可（``backends/__init__.py`` 里统一 import）。
|
||||
"""
|
||||
|
||||
from typing import Union
|
||||
|
||||
from .base import AsyncStorageBackend, StorageBackend
|
||||
from .base import AsyncStorageBackend
|
||||
from .exceptions import StorageConfigError
|
||||
|
||||
BackendClass = Union[type[StorageBackend], type[AsyncStorageBackend]]
|
||||
BackendClass = type[AsyncStorageBackend]
|
||||
|
||||
_REGISTRY: dict[tuple[str, str], BackendClass] = {}
|
||||
|
||||
VALID_MODES = ("sync", "async")
|
||||
_REGISTRY: dict[str, BackendClass] = {}
|
||||
|
||||
|
||||
def _check_mode(mode: str) -> None:
|
||||
if mode not in VALID_MODES:
|
||||
raise StorageConfigError(f"不支持的 mode: {mode!r}，可选值: {VALID_MODES}")
|
||||
|
||||
|
||||
def register_backend(name: str, mode: str = "sync"):
|
||||
"""类装饰器：把一个后端类注册为 (mode, name) 对应的实现。"""
|
||||
_check_mode(mode)
|
||||
def register_backend(name: str):
|
||||
"""类装饰器：把一个后端类注册为 ``name`` 对应的实现。"""
|
||||
|
||||
def _decorator(cls: BackendClass) -> BackendClass:
|
||||
key = (mode, name)
|
||||
if key in _REGISTRY and _REGISTRY[key] is not cls:
|
||||
if name in _REGISTRY and _REGISTRY[name] is not cls:
|
||||
raise StorageConfigError(
|
||||
f"存储后端 (mode={mode}, type={name}) 已被注册为 {_REGISTRY[key]!r}"
|
||||
f"存储后端 (type={name}) 已被注册为 {_REGISTRY[name]!r}"
|
||||
)
|
||||
_REGISTRY[key] = cls
|
||||
_REGISTRY[name] = cls
|
||||
return cls
|
||||
|
||||
return _decorator
|
||||
|
||||
|
||||
def get_backend_class(name: str, mode: str = "sync") -> BackendClass:
|
||||
_check_mode(mode)
|
||||
key = (mode, name)
|
||||
def get_backend_class(name: str) -> BackendClass:
|
||||
try:
|
||||
return _REGISTRY[key]
|
||||
return _REGISTRY[name]
|
||||
except KeyError:
|
||||
available = ", ".join(
|
||||
f"{m}:{n}" for (m, n) in sorted(_REGISTRY)
|
||||
) or "(无)"
|
||||
available = ", ".join(sorted(_REGISTRY)) or "(无)"
|
||||
raise StorageConfigError(
|
||||
f"未知的存储后端 (mode={mode}, type={name})，当前已注册: {available}"
|
||||
f"未知的存储后端 (type={name})，当前已注册: {available}"
|
||||
)
|
||||
|
||||
|
||||
def registered_backends() -> dict[tuple[str, str], BackendClass]:
|
||||
def registered_backends() -> dict[str, BackendClass]:
|
||||
return dict(_REGISTRY)
|
||||
|
||||
@@ -210,7 +210,6 @@ def build_object_store(bucket_name: str | None = None) -> Any:
|
||||
return create_storage(
|
||||
{
|
||||
"type": "local",
|
||||
"mode": "async",
|
||||
"base_dir": str(
|
||||
Path(settings.local_storage_base_dir) / "version"
|
||||
),
|
||||
@@ -219,7 +218,6 @@ def build_object_store(bucket_name: str | None = None) -> Any:
|
||||
return create_storage(
|
||||
{
|
||||
"type": "s3",
|
||||
"mode": "async",
|
||||
"bucket": bucket_name or settings.s3_version_bucket,
|
||||
"endpoint_url": settings.s3_endpoint,
|
||||
"aws_access_key_id": settings.s3_access_key,
|
||||
|
||||
Reference in New Issue
Block a user