diff --git a/backend/src/backend/services/storage.py b/backend/src/backend/services/storage.py index 9741123..00c5452 100644 --- a/backend/src/backend/services/storage.py +++ b/backend/src/backend/services/storage.py @@ -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": { diff --git a/common/pyproject.toml b/common/pyproject.toml index 8c1e319..1a1068f 100644 --- a/common/pyproject.toml +++ b/common/pyproject.toml @@ -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", diff --git a/common/src/common/config.py b/common/src/common/config.py index 93cf9ee..f88068f 100644 --- a/common/src/common/config.py +++ b/common/src/common/config.py @@ -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", diff --git a/common/src/common/storage/__init__.py b/common/src/common/storage/__init__.py index 4a4906d..53a0ad9 100644 --- a/common/src/common/storage/__init__.py +++ b/common/src/common/storage/__init__.py @@ -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", diff --git a/common/src/common/storage/backends/local.py b/common/src/common/storage/backends/local.py index 9c2785e..d81a768 100644 --- a/common/src/common/storage/backends/local.py +++ b/common/src/common/storage/backends/local.py @@ -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() diff --git a/common/src/common/storage/backends/s3.py b/common/src/common/storage/backends/s3.py index 6f067a2..0d5a69d 100644 --- a/common/src/common/storage/backends/s3.py +++ b/common/src/common/storage/backends/s3.py @@ -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 @@ -418,4 +269,4 @@ class S3AsyncStorageBackend(AsyncStorageBackend): 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 await self.stat(dst_key) + return await self.stat(dst_key) \ No newline at end of file diff --git a/common/src/common/storage/base.py b/common/src/common/storage/base.py index 30b40d3..0ecf871 100644 --- a/common/src/common/storage/base.py +++ b/common/src/common/storage/base.py @@ -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): """异步存储后端统一抽象基类。""" diff --git a/common/src/common/storage/example_usage.py b/common/src/common/storage/example_usage.py deleted file mode 100644 index 4c2d06a..0000000 --- a/common/src/common/storage/example_usage.py +++ /dev/null @@ -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() diff --git a/common/src/common/storage/factory.py b/common/src/common/storage/factory.py index 1b152c1..2529e6a 100644 --- a/common/src/common/storage/factory.py +++ b/common/src/common/storage/factory.py @@ -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, diff --git a/common/src/common/storage/registry.py b/common/src/common/storage/registry.py index 2d061cb..0748b59 100644 --- a/common/src/common/storage/registry.py +++ b/common/src/common/storage/registry.py @@ -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) diff --git a/schedule/src/schedule/application/service.py b/schedule/src/schedule/application/service.py index 61bcc00..967cffb 100644 --- a/schedule/src/schedule/application/service.py +++ b/schedule/src/schedule/application/service.py @@ -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,