From 5cbbe8213aca22607bb8a161346d4eb05011b0d9 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Fri, 31 Jul 2026 13:34:33 +0800 Subject: [PATCH] review --- .gitignore | 6 +- README.md | 219 ++++++++++++++++++++------ backend/src/backend/storage_api.py | 20 +-- backend/src/backend/storage_client.py | 67 +++++++- 4 files changed, 252 insertions(+), 60 deletions(-) diff --git a/.gitignore b/.gitignore index 388de3b..eaf4562 100644 --- a/.gitignore +++ b/.gitignore @@ -2,7 +2,9 @@ .env -.idea - .venv +__pycache__/ +*.pyc +.DS_Store + diff --git a/README.md b/README.md index 85dbcc9..746eefe 100644 --- a/README.md +++ b/README.md @@ -1,69 +1,196 @@ -# Model Platform Refactored +# Model Platform -本项目是简化后的模型实验开发平台：React Router 前端、FastAPI Backend、 -Runtime、Schedule Executor、MySQL、RustFS、共享 Jupyter 与 Nginx Gateway。 -Redis 已移除。 +A self-hosted **Jupyter-based model development platform** that combines +an interactive workspace, a DAG scheduler, an object-storage-backed artifact +store, and per-workspace runtime isolation — all behind a single Nginx +gateway. -## 目录 +> Stack: React Router SPA · FastAPI · APScheduler · MySQL · RustFS (S3) +> · shared Jupyter · FUSE mount via rclone +> Single ingress (Nginx :80); all other services are Docker-internal. + +## What it does + +| Capability | Where | +|---|---| +| Workspace-scoped notebook editing with row-level lock | `backend/jupyter.py` + `scripts.is_locked` | +| Authenticated Jupyter routing (browser never sees the runtime token) | `nginx/default.conf` + `auth_request` + `backend/jupyter.py` | +| Object storage for notebooks / scripts / versions / run logs (RustFS, S3 API) | `common/storage/` + `backend/scripts.py` | +| DAG-style scheduling: nodes, edges, cron, manual trigger, retries, snapshots | `backend/schedules.py` + `backend/schedule_runs.py` + `schedule/` (5 modules) | +| DAG execution via MySQL Outbox (no Redis, no in-process queues) | `schedule/orchestrator.py` + `schedule/worker.py` | +| Per-workspace Jupyter sub-process pool with asyncio locks | `runtime/process.py` | +| rclone FUSE mount of the workspace bucket into the runtime | `runtime/mount.py` | +| MySQL-only persistence (26 tables, soft-delete, no foreign keys) | `common/db/models/` | + +## Architecture at a glance + +``` + ┌────────────────────┐ + │ Browser (SPA) │ + └─────────┬──────────┘ + │ HTTPS / WS + ┌─────────▼──────────┐ + │ Nginx (only :80) │ ← templates/default.conf + │ /api/ /jupyter/ /storage/ + └────┬───────┬──────┘ + │ │ + ┌──────────────┘ └─────────────┐ + ▼ ▼ + ┌──────────────────┐ ┌──────────────────────┐ + │ FastAPI Backend │ │ Runtime (Jupyter) │ + │ + /internal/v1 │ control │ - rclone FUSE mount │ + │ (storage) ├──────────────►│ - subprocess pool │ + │ - DAG CRUD │ │ (per workspace) │ + │ - script CRUD │ └──────────┬───────────┘ + │ - auth_request │ │ FUSE + │ - /api/v1/... │ ▼ + └────┬──────┬──────┘ ┌──────────────────────┐ + │ │ │ RustFS (S3) │ + │ └──────── HTTP ───────►│ bucket: workspaces │ + ▼ │ bucket: versions │ + ┌────────────┐ │ bucket: run-logs │ + │ MySQL │◄───────── poll ─────│ │ + │ - 26 tbls │ └──────────────────────┘ + │ - outbox │ + │ - jobstore │ + └────┬───────┘ + ▲ + │ outbox poll + ┌────┴──────────────────────────┐ + │ Schedule Executor │ + │ - CronScheduler (APScheduler) │ + │ - DispatchOrchestrator │ + │ - NodeExecutor (worker) │ + │ - SchedulerService (facade) │ + └───────────────────────────────┘ +``` + +Detailed design lives in `ARCHITECTURE.md`. Implementation deviations and +recent refactors are recorded in `HANDOVER.md`. + +## Repository layout ```text -frontend/app/ - components/ 通用组件 - features/platform/ 平台主框架与脚本工作台 - features/schedules/ 调度画布 - features/admin/ 工作台与系统管理 - routes/ React Router 路由 - services/ API 调用与 DTO -backend/ 核心业务 API -runtime/ Jupyter 与文件编辑租约 -schedule/ APScheduler + DAG Executor -common/ 数据库模型和公共能力 -nginx/ 前端静态站点与统一反向代理 +frontend/ React Router SPA +backend/ FastAPI: public API + internal storage API +runtime/ Jupyter subprocess manager + rclone FUSE +schedule/ DAG scheduler (5 modules: context/scheduler/ + orchestrator/worker/service) +common/ Settings, SQLAlchemy models, storage SDK, + outbox events, jobstore +migrations/ Alembic baseline + per-feature revisions +nginx/ (concept only — see "Container" below) +scripts/ nginx-entrypoint.sh (template renderer) +docker-compose.yml 4 services — web / backend / runtime / schedule +default.conf Nginx template (mounted, rendered at start) +.env.example All env vars consumed by common.config.Settings ``` -## 简化后的调用链 +## Containers -```text -Browser -> Nginx Gateway -> FastAPI Backend -> MySQL / RustFS - -> Runtime -> shared Jupyter -Backend --HTTP push-----> Schedule Executor -Schedule Executor --poll MySQL Outbox / APScheduler--> execute DAG -``` +| Service | Image | Exposed | Purpose | +|---|---|---|---| +| `web` | `nginx:alpine` | host `:8888` → `:80` | SPA, `/api/` reverse-proxy, `/jupyter/{ws}/` auth_request proxy, `/storage/` RustFS passthrough | +| `backend` | `Dockerfile` | internal only | DAG CRUD, script CRUD, schedule triggers, `/api/v1/auth/jupyter`, `/internal/v1/*` storage control plane | +| `runtime` | `Dockerfile` | internal only | Per-workspace Jupyter sub-process pool, rclone FUSE mount of `workspaces` bucket | +| `schedule` | `Dockerfile` | internal only | Cron tick + DAG execution via MySQL Outbox polling | -## 启动 +The architecture **deliberately has only one host port** (the gateway); +all other services are on the Docker internal network. This is enforced in +`docker-compose.yml` — no `ports:` on backend / runtime / schedule. -Windows PowerShell： - -```powershell -Copy-Item .env.example .env -docker compose config -docker compose up -d --build -docker compose ps -``` - -Git Bash/Linux： +## Quick start ```bash cp .env.example .env -docker compose config +# Edit .env — at minimum change MYSQL password and RUSTFS credentials. + +# Static check +uv sync --all-packages +uv run --frozen --package backend python -m compileall -q backend/src common/src +uv run --frozen --package schedule python -m compileall -q schedule/src +uv run --frozen --package runtime python -m compileall -q runtime/src + +# Apply schema +uv run --frozen --package backend alembic upgrade head + +# Bring up the stack +docker compose config # validate docker compose up -d --build docker compose ps ``` -访问：`http://127.0.0.1:8081`。 +Visit `http://localhost:8888`. -查看日志： +### Logs ```bash -docker compose logs -f backend runtime schedule gateway +docker compose logs -f backend +docker compose logs -f schedule +docker compose logs -f runtime ``` -数据库迁移由 `migrate` 一次性容器在 Backend 启动前自动执行。 +### Tear down (keeps MySQL + RustFS volumes) -## Windows 快速脚本 - -```powershell -.\scripts\start.ps1 -.\scripts\logs.ps1 -.\scripts\stop.ps1 +```bash +docker compose down ``` + +### Wipe data + +```bash +docker compose down -v +``` + +## Configuration + +All environment variables are declared once in `common/src/common/config.py` +as a pydantic-settings `Settings` class, with a `@lru_cache` singleton. +Adding a new env var: + +1. Add the field to `Settings` in `common/src/common/config.py` (with a + sensible default so dev-env "just works"). +2. Add the line to `.env.example` with a comment. +3. Use `settings.` at the call site. Never `os.environ["..."]`. + +See `DEVELOP.md` for the full list of variables and their meanings. + +## Storage layout + +Single bucket `workspaces` (configurable via `RUSTFS_WORKSPACE_BUCKET`). +The object key is a flat two-level path — `workspace_id` and a server- +issued `ulid` for the object. The file name, extension, content type, +and logical path live in the `StorageObjects` and `Scripts` rows, not in +the S3 key, so the bucket can be re-organised without a database +rewrite. + +``` +s3://workspaces/ +└── / + ├── # script / notebook / data resource + ├── + └── ... + +s3://versions/ (RUSTFS_VERSION_BUCKET — reserved, used by publish_version) +s3://run-logs/ (RUSTFS_RUN_LOG_BUCKET — reserved, used by node executor) +``` + +To find the original file name and its logical path for a given bucket +object, join `StorageObjects.bucket_name + object_key` to the row. + +Backend code never writes to the container's local filesystem. Schedule +Executor stages node artifacts in `tempfile.TemporaryDirectory()` (auto- +cleaned). Only the `runtime` container keeps a host volume — it is required +by the rclone FUSE mount. + +## Documentation + +- `README.md` (this file) — quick orientation +- `ARCHITECTURE.md` — design diagrams + simplification history +- `HANDOVER.md` — implementation deviations, recent refactors, pending work +- `DEVELOP.md` — developer guide (env vars, code conventions, common tasks) +- `CLAUDE.md` — agent-facing conventions for the repo + +## License + +Internal. diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index ff97327..c49261b 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -165,17 +165,15 @@ async def create_upload_record( upload = existing else: upload_id = new_ulid() - file_name = safe_file_name(payload.file_name) bucket_name = ( workspace.artifact_bucket or request.app.state.default_bucket ) - # Default layout: one top-level folder per workspace inside the - # ``workspaces`` bucket. ``bucket_name`` already encodes the workspace - # namespace, so the key starts with the workspace id directly. - object_key = ( - f"{payload.workspace_id}/" - f"{payload.usage_type}/{upload_id}/{file_name}" - ) + # Object key is a flat two-level path: workspace id + upload id. The + # original file name and content type live in the StorageObjects row + # (file_name / mime_type / object_key) — they are not part of the + # key itself, so the bucket can be reorganised without rewriting + # the database. + object_key = f"{payload.workspace_id}/{upload_id}" upload = UploadSessions( upload_id=upload_id, workspace_id=payload.workspace_id, @@ -320,7 +318,11 @@ async def complete_upload_record( status.HTTP_409_CONFLICT, "uploaded object hash does not match expected_hash") - file_name = upload.object_key.rsplit("/", 1)[-1] + # The object key is just ``{workspace_id}/{ulid}`` — it does not encode + # the file name. Use the original file name from the upload session + # (carried via payload.file_name) so the StorageObjects row still + # records the user-visible name + extension. + file_name = safe_file_name(payload.file_name) item = StorageObjects( storage_object_id=new_ulid(), workspace_id=upload.workspace_id, diff --git a/backend/src/backend/storage_client.py b/backend/src/backend/storage_client.py index 96f41e4..bf924d3 100644 --- a/backend/src/backend/storage_client.py +++ b/backend/src/backend/storage_client.py @@ -1,5 +1,66 @@ -"""Backward-compatible re-export of the shared storage client.""" +"""Backend-bound storage client. -from common.storage.client import StorageClient +Re-exports :class:`StorageClient` under the same name used by callers in +``backend/``. The default client raises :class:`StorageClientError` from +``common.storage.client`` so it stays usable from non-FastAPI contexts. +Inside FastAPI route handlers we want HTTP-shaped errors, so this module +also exposes :class:`BackendStorageClient`, a thin wrapper that translates +the framework-agnostic errors into ``HTTPException``. +""" -__all__ = ["StorageClient"] +from __future__ import annotations + +from typing import Any + +from fastapi import HTTPException, status + +from common.storage.client import ( + StorageClient, + StorageClientError, + StorageRequestFailed, + StorageUnavailable, +) + + +__all__ = ["BackendStorageClient", "StorageClient", "StorageClientError"] + + +def _to_http_exception(exc: StorageClientError) -> HTTPException: + if isinstance(exc, StorageUnavailable): + return HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + { + "code": "STORAGE_UNAVAILABLE", + "message": "Storage service temporarily unavailable", + "retryable": True, + "details": {}, + }, + ) + if isinstance(exc, StorageRequestFailed): + return HTTPException(exc.status_code, exc.detail) + return HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + "Storage client error", + ) + + +class BackendStorageClient(StorageClient): + """Storage client that raises ``HTTPException`` for web callers.""" + + async def _request( + self, + method: str, + path: str, + *, + payload: dict[str, Any] | None = None, + ) -> dict[str, Any]: + try: + return await super()._request(method, path, payload=payload) + except StorageClientError as exc: + raise _to_http_exception(exc) from exc + + +# Re-bind the imported symbol so existing backend call sites that import +# ``StorageClient`` from this module transparently get the FastAPI-bound +# variant without changing every import statement. +StorageClient = BackendStorageClient