feat: rclone client

backend/src/backend/rclone_rc_client.py

@@ -0,0 +1,92 @@
+"""rclone RC client.
+
+Used by the backend to invalidate the rclone FUSE directory cache after
+writing a new workspace object (notebook / script / upload). The runtime
+container already starts rclone with ``--rc --rc-addr 0.0.0.0:5572
+--rc-no-auth`` (see ``runtime/src/runtime/mount.py``), so we only need
+an HTTP client here — no extra runtime plumbing.
+
+Failure is logged and swallowed: the VFS refresh is best-effort and
+must never bubble up to the API caller.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+from loguru import logger
+
+
+@dataclass(frozen=True)
+class RcloneRCError(Exception):
+    status_code: int
+    detail: Any
+
+
+_RCLONE_RC_TRANSPORT_ERROR = RcloneRCError(
+    503,
+    {
+        "code": "RCLONE_RC_UNAVAILABLE",
+        "message": "rclone RC 暂时不可用",
+        "retryable": True,
+        "details": {},
+    },
+)
+
+
+class RcloneRCClient:
+    def __init__(self, client: httpx.AsyncClient) -> None:
+        self.client = client
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        try:
+            response = await self.client.request(method, path, json=payload)
+        except httpx.RequestError as exc:
+            raise _RCLONE_RC_TRANSPORT_ERROR from exc
+        if response.is_error:
+            try:
+                detail = response.json().get("detail", response.text)
+            except ValueError:
+                detail = response.text
+            raise RcloneRCError(response.status_code, detail)
+        return response.json()
+
+    async def vfs_refresh(
+        self,
+        dir_path: str,
+        *,
+        recursive: bool = True,
+    ) -> None:
+        """Invalidate the FUSE dir-cache for ``dir_path``.
+
+        Fires ``POST /vfs/refresh`` against the rclone RC server with
+        ``_async=true`` so the call returns immediately while rclone
+        performs the directory walk in the background. Errors are
+        logged and swallowed — a refresh failure must never fail the
+        API call that triggered it.
+        """
+        try:
+            await self._request(
+                "POST",
+                "/vfs/refresh",
+                {
+                    "dir": dir_path,
+                    "recursive": recursive,
+                    "_async": True,
+                },
+            )
+            logger.info(
+                f"vfs_refresh dir={dir_path} recursive={recursive} ok"
+            )
+        except Exception as exc:  # noqa: BLE001 - best-effort
+            logger.warning(f"vfs_refresh dir={dir_path} failed: {exc}")
+
+
+__all__ = ["RcloneRCClient", "RcloneRCError"]