modify: S3对象存储

README.md

@@ -68,6 +68,25 @@ class MyPlugin(Plugin):
 #     url = "postgresql+asyncpg://user:pwd@127.0.0.1:5432/mydb"
 #     generate_schemas = true
 
+# 对象存储接入（可选）：在插件 config.toml 配置 storage 段后，
+# 基类 init_storage() 自动初始化 self.storage（S3/Minio，直连不经网关）::
+#
+#     # config/workers/my_plugin/config.toml
+#     [workers.config.storage]
+#     endpoint = "192.168.1.10:9000"   # Minio 地址 host:port（不带协议前缀）
+#     access_key = "minioadmin"
+#     secret_key = "minioadmin"
+#     bucket = "chongming-files"
+#     secure = false
+#     # region = ""                    # 可选
+#     # auto_create_bucket = true      # 默认 true：启动时确保 bucket 存在
+#
+# 语义：未配置 / enabled=false → self.storage 为 None（插件自行降级）；
+# 缺必填键（endpoint/access_key/secret_key/bucket）→ 启动硬失败（ValueError）；
+# 连接失败 → 告警降级为 None。API 见 chongming_worker.storage.StorageClient
+# （put_file/put_bytes/get_file/get_bytes/delete/presigned_put_url/presigned_get_url
+# 及对应 async a* 版本，handler 中直接 await）。
+
 # 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法一致）
 plugin = MyPlugin()
 create_workers = plugin.create_workers
@@ -89,6 +108,7 @@ cleanup = plugin.cleanup
 - **幂等**：重复 `create_workers` 只初始化一次（`asyncio.Lock` 守卫，并发调用安全）
 - **统一日志**：`logger` 自动带 `[type]` 前缀
 - **数据库**：`init_database()` 统一建库建表、记录，`cleanup` 自动 `dispose()`（SQLite 按 `db_name`；config 含 `database.url` 时走 PostgreSQL）
+- **对象存储**：`init_storage()` 按 config 的 `storage` 段初始化 `self.storage`（S3/Minio，可选），`cleanup` 自动关闭
 - **global_var**：`set_var()` / `get_var()` 统一读写插件 `global_var` 模块；`cleanup` 时自动将已发布的变量重置为 `None`，避免业务代码拿到已释放的资源
 - **后台任务**：`spawn_task()` 跟踪，`cleanup` 时自动取消并 `gather`
 - **清理钩子**：`register_cleanup()` 注册，`cleanup` 时逆序执行

pyproject.toml

@@ -4,7 +4,7 @@ requires = ["hatchling"]
 
 [project]
 name = "chongming-worker"
-version = "0.2.0"
+version = "0.3.0"
 description = "chongming worker"
 readme = "README.md"
 requires-python = ">=3.12"
@@ -12,6 +12,7 @@ dependencies = [
     "eclipse-zenoh>=1.9.0",
     "pydantic>=2.0",
     "tomli>=2.4.1",
+    "minio>=7.2.0",
 ]
 
 [project.scripts]

src/chongming_worker/__init__.py

@@ -25,6 +25,9 @@ from chongming_worker.runtime import (
 from chongming_worker.plugin import (
     Plugin,
 )
+from chongming_worker.storage import (
+    StorageClient,
+)
 
 __all__ = [
     "RPCMethod",
@@ -41,4 +44,5 @@ __all__ = [
     "create_worker",
     "create_workers",
     "Plugin",
+    "StorageClient",
 ]

src/chongming_worker/plugin.py

@@ -50,6 +50,8 @@ Plugin 基类 — 统一 Worker 插件初始化与清理生命周期
   - 统一日志：``logger`` 自动带 ``[type]`` 前缀
   - 数据库：``_bootstrap()`` / ``init_database()`` 统一建库建表、记录，``cleanup`` 自动 dispose
     （SQLite 按 ``db_name`` 建库；config 含 ``database.url`` 时优先走 PostgreSQL）
+  - 对象存储：``_bootstrap()`` / ``init_storage()`` 按 config 的 storage 段初始化
+    （S3/Minio，``[workers.config.storage]``），``cleanup`` 自动关闭
   - global_var：``set_var()`` 写入插件自己的 ``global_var`` 模块（白名单由插件决定）
   - 后台任务：``spawn_task()`` 跟踪，``cleanup`` 时自动取消
   - 清理钩子：``register_cleanup()`` 注册，``cleanup`` 时逆序执行
@@ -90,6 +92,9 @@ class Plugin:
         self.config: dict = {}
         #: Database 实例（由 init_database 填充；类型标注为 Any 以避免强依赖 chongming_database）
         self.db: Any = None
+        #: StorageClient 实例（由 init_storage 按 config 的 storage 段填充；类型标注为 Any
+        #: 以避免强依赖 minio）
+        self.storage: Any = None
         self._initialized = False
         #: 初始化互斥锁——同一进程内并发 create_workers（如测试模拟多 worker 启动）串行化
         self._init_lock = asyncio.Lock()
@@ -104,12 +109,14 @@ class Plugin:
     # ------------------------------------------------------------------
     async def _bootstrap(self) -> None:
         """
-        默认资源引导：按 db_name（SQLite）或 database.url（PostgreSQL）初始化数据库。
+        默认资源引导：按 db_name（SQLite）或 database.url（PostgreSQL）初始化数据库，
+        并按 config 的 storage 段初始化对象存储（S3/Minio，可选）。
 
-        由基类在 on_init 之前自动调用；子类需要完全自定义建库逻辑时覆写此方法。
+        由基类在 on_init 之前自动调用；子类需要完全自定义引导逻辑时覆写此方法。
         """
         if self.db_name or self._database_config().get("url"):
             await self.init_database(self.db_name)
+        await self.init_storage()
 
     async def on_init(self) -> None:
         """首次 create_workers 时调用一次，初始化业务资源（数据库已由 _bootstrap 引导）。"""
@@ -183,6 +190,13 @@ class Plugin:
             if self.db is not None:
                 await self.db.dispose()
                 self.db = None
+            # 4.5 关闭对象存储客户端（S3/Minio，可选）
+            if self.storage is not None:
+                try:
+                    self.storage.close()
+                except Exception as e:  # noqa: BLE001
+                    self.logger.warning("[%s] 关闭对象存储失败: %s", self.type, e)
+                self.storage = None
             # 5. 清空通过 set_var 发布的全局变量，避免 global_var 继续持有已释放资源
             for name in list(self._vars):
                 try:
@@ -247,6 +261,33 @@ class Plugin:
         self.logger.info("[%s] 数据库已初始化: %s", self.type, db_path)
         return db
 
+    async def init_storage(self) -> Any:
+        """
+        初始化对象存储客户端并存入 self.storage（S3/Minio，配置驱动，可选）。
+
+        读取运行时配置中的 storage 段（插件 config.toml 的 ``[workers.config.storage]``：
+        endpoint/access_key/secret_key/bucket/secure/region/auto_create_bucket）。
+
+        语义（详见 ``chongming_worker.storage.StorageClient.from_config``）:
+          - 未配置或 ``enabled = false`` → self.storage 保持 None（插件自行降级）
+          - 配置段存在但缺必填键 → 抛 ValueError（启动硬失败，快速反馈配置错误）
+          - 连接失败 / bucket 不可用 → 告警并降级为 None
+
+        返回 StorageClient 或 None。
+        """
+        from chongming_worker.storage import StorageClient
+
+        storage_conf = self.config.get("storage") if isinstance(self.config, dict) else None
+        self.storage = StorageClient.from_config(
+            storage_conf if isinstance(storage_conf, dict) else None
+        )
+        if self.storage is not None:
+            self.logger.info(
+                "[%s] 对象存储已就绪: %s bucket=%s",
+                self.type, self.storage.endpoint, self.storage.bucket,
+            )
+        return self.storage
+
     def set_var(self, name: str, value: Any) -> None:
         """
         写入运行时变量，并同步到插件 global_var 模块。

src/chongming_worker/storage.py

@@ -0,0 +1,210 @@
+"""
+对象存储客户端（S3/Minio）— 配置驱动，进程插件直连，不依赖网关
+
+进程插件（external worker）在 ``config/workers/<name>/config.toml`` 配置
+``[workers.config.storage]`` 段后，由 Plugin 基类 ``_bootstrap`` 自动初始化
+``self.storage``（StorageClient）。上传 / 下载 / 预签名 URL 直接与 Minio
+交互，数据路径完全不经过网关（预签名 URL 模式下网关甚至不参与数据传输）。
+
+配置示例::
+
+    # config/workers/<name>/config.toml
+    [workers.config.storage]
+    endpoint = "192.168.1.10:9000"   # Minio 地址 host:port（不带协议前缀）
+    access_key = "minioadmin"
+    secret_key = "minioadmin"
+    bucket = "chongming-files"
+    secure = false                   # false=http（默认），true=https
+    region = ""                      # 可选
+    auto_create_bucket = true        # 可选，默认 true：启动时确保 bucket 存在
+
+``StorageClient.from_config`` 语义:
+  - 无配置段 / ``enabled = false`` → 返回 None（静默关闭）
+  - 配置段存在但缺必填键（endpoint/access_key/secret_key/bucket）→ 抛 ValueError
+    （启动硬失败，与契约校验哲学一致，快速反馈配置错误）
+  - 连接失败 / bucket 不可用 → warning 日志 + 返回 None（降级，与 mqtt_forward 一致）
+"""
+from __future__ import annotations
+
+import asyncio
+import io
+import logging
+from datetime import timedelta
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger("chongming_worker.storage")
+
+#: 必填配置键（缺失时启动硬失败）
+_REQUIRED_KEYS = ("endpoint", "access_key", "secret_key", "bucket")
+
+
+class StorageClient:
+    """Minio(S3) 客户端包装：同步核心 + async 便捷方法。
+
+    minio SDK 为 urllib3 同步实现，async 方法用 ``asyncio.to_thread`` 包装，
+    供 RPC handler 直接 ``await``；同步方法供后台任务 / 脚本直接使用。
+    """
+
+    def __init__(self, client: Any, bucket: str, endpoint: str) -> None:
+        self._client = client
+        self.bucket = bucket
+        self.endpoint = endpoint
+
+    # ------------------------------------------------------------------
+    # 构造
+    # ------------------------------------------------------------------
+    @classmethod
+    def from_config(cls, storage_conf: dict | None) -> "StorageClient | None":
+        """按配置构造客户端；未配置 / 禁用返回 None，配置错误抛 ValueError。
+
+        Args:
+            storage_conf: ``[workers.config.storage]`` 段展平后的 dict，
+                None / 空 dict 表示未启用（静默关闭）
+
+        Returns:
+            StorageClient；禁用或连接失败（降级）时为 None
+
+        Raises:
+            ValueError: 配置段存在但缺少必填键（endpoint/access_key/secret_key/bucket）
+        """
+        if not storage_conf or not isinstance(storage_conf, dict):
+            return None
+        if not storage_conf.get("enabled", True):
+            logger.info("storage 已禁用（enabled=false），跳过初始化")
+            return None
+
+        missing = [k for k in _REQUIRED_KEYS if not storage_conf.get(k)]
+        if missing:
+            raise ValueError(f"storage 配置缺少必填项: {', '.join(missing)}")
+
+        try:
+            from minio import Minio
+            import urllib3
+        except ImportError as e:  # pragma: no cover - 依赖缺失属打包错误
+            raise RuntimeError(
+                "缺少 minio 依赖：请确认 chongming-worker 的 minio 依赖已安装"
+            ) from e
+
+        endpoint = storage_conf["endpoint"]
+        # 自定义连接池：连接超时 3s（Minio 不可达时快速降级，避免 SDK 默认
+        # 长重试拖慢 worker 启动），读超时 60s（大文件传输留足余量）
+        http_client = urllib3.PoolManager(
+            timeout=urllib3.Timeout(connect=3.0, read=60.0)
+        )
+        client = Minio(
+            endpoint,
+            access_key=storage_conf["access_key"],
+            secret_key=storage_conf["secret_key"],
+            secure=bool(storage_conf.get("secure", False)),
+            region=storage_conf.get("region") or None,
+            http_client=http_client,
+        )
+        return cls._connect(client, endpoint, storage_conf)
+
+    @classmethod
+    def _connect(
+        cls, client: Any, endpoint: str, storage_conf: dict
+    ) -> "StorageClient | None":
+        """连接校验 + bucket 确保；失败告警并降级为 None。"""
+        bucket = storage_conf["bucket"]
+        try:
+            exists = client.bucket_exists(bucket)
+            if not exists:
+                if storage_conf.get("auto_create_bucket", True):
+                    client.make_bucket(bucket, location=storage_conf.get("region") or None)
+                else:
+                    raise RuntimeError(f"bucket 不存在: {bucket}")
+            return cls(client, bucket, endpoint)
+        except Exception as e:  # noqa: BLE001
+            logger.warning(
+                "storage 连接失败，降级为不可用: %s (endpoint=%s, bucket=%s)",
+                e, endpoint, bucket,
+            )
+            return None
+
+    # ------------------------------------------------------------------
+    # 同步核心（minio SDK 为同步实现）
+    # ------------------------------------------------------------------
+    def put_file(
+        self, object_key: str, local_path: str | Path, content_type: str | None = None
+    ) -> str:
+        """上传本地文件，返回 object_key。"""
+        self._client.fput_object(
+            self.bucket, object_key, str(local_path), content_type=content_type
+        )
+        return object_key
+
+    def put_bytes(
+        self, object_key: str, data: bytes, content_type: str | None = None
+    ) -> str:
+        """上传内存字节，返回 object_key。"""
+        self._client.put_object(
+            self.bucket, object_key, io.BytesIO(data), len(data),
+            content_type=content_type,
+        )
+        return object_key
+
+    def get_file(self, object_key: str, local_path: str | Path) -> Path:
+        """下载到本地文件，返回目标路径。"""
+        self._client.fget_object(self.bucket, object_key, str(local_path))
+        return Path(local_path)
+
+    def get_bytes(self, object_key: str) -> bytes:
+        """下载为内存字节。"""
+        resp = self._client.get_object(self.bucket, object_key)
+        try:
+            return resp.read()
+        finally:
+            resp.close()
+            resp.release_conn()
+
+    def delete(self, object_key: str) -> None:
+        """删除对象（不存在时由服务端决定，一般静默成功）。"""
+        self._client.remove_object(self.bucket, object_key)
+
+    def presigned_put_url(self, object_key: str, expires: int = 3600) -> str:
+        """预签名 PUT URL：拿到 URL 者可直传 Minio（数据不经网关）。"""
+        return self._client.presigned_put_object(
+            self.bucket, object_key, expires=timedelta(seconds=expires)
+        )
+
+    def presigned_get_url(self, object_key: str, expires: int = 3600) -> str:
+        """预签名 GET URL：拿到 URL 者可直接下载（数据不经网关）。"""
+        return self._client.presigned_get_object(
+            self.bucket, object_key, expires=timedelta(seconds=expires)
+        )
+
+    def close(self) -> None:
+        """兼容清理钩子（minio SDK 无显式关闭；urllib3 连接池随解释器回收）。"""
+
+    # ------------------------------------------------------------------
+    # async 便捷方法（to_thread 包装同步 SDK，RPC handler 直接 await）
+    # ------------------------------------------------------------------
+    async def aput_file(
+        self, object_key: str, local_path: str | Path, content_type: str | None = None
+    ) -> str:
+        return await asyncio.to_thread(self.put_file, object_key, local_path, content_type)
+
+    async def aput_bytes(
+        self, object_key: str, data: bytes, content_type: str | None = None
+    ) -> str:
+        return await asyncio.to_thread(self.put_bytes, object_key, data, content_type)
+
+    async def aget_file(self, object_key: str, local_path: str | Path) -> Path:
+        return await asyncio.to_thread(self.get_file, object_key, local_path)
+
+    async def aget_bytes(self, object_key: str) -> bytes:
+        return await asyncio.to_thread(self.get_bytes, object_key)
+
+    async def adelete(self, object_key: str) -> None:
+        await asyncio.to_thread(self.delete, object_key)
+
+    async def apresigned_put_url(self, object_key: str, expires: int = 3600) -> str:
+        return await asyncio.to_thread(self.presigned_put_url, object_key, expires)
+
+    async def apresigned_get_url(self, object_key: str, expires: int = 3600) -> str:
+        return await asyncio.to_thread(self.presigned_get_url, object_key, expires)
+
+
+__all__ = ["StorageClient"]