tao.chen/model-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' of http://8.153.151.51:8888/team_group/model-develop into develop

Browse Source 
# Conflicts:
#	schedule/src/schedule/orchestrator.py
#	schedule/src/schedule/worker.py
This commit is contained in:
Winnie
2026-08-04 14:41:22 +08:00
parent a2deae2f22 1cf2eecbc9
commit f3cc83c5fa
25 changed files with 743 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+7
View File
@@ -45,3 +45,10 @@ RUSTFS_VERSION_BUCKET=versions
RUSTFS_RUN_LOG_BUCKET=run-logs
RUSTFS_TRASH_BUCKET=trash
RUSTFS_TRASH_RETENTION_DAYS=30
# rclone RC (HTTP control API). The runtime container starts rclone with
# `--rc --rc-addr 0.0.0.0:5572 --rc-no-auth` (see runtime/src/runtime/mount.py),
# so the backend can POST /vfs/refresh here to invalidate the FUSE dir-cache
# after writing new workspace files. Default points at the runtime service
# over the compose network.
RCLONE_RC_URL=http://runtime:5572
backend/Dockerfile
+6 -1
View File
@@ -3,7 +3,8 @@ FROM python:3.12-slim-bookworm
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PYTHONPATH=/app \
PATH="/app/.venv/bin:${PATH}"
PATH="/app/.venv/bin:${PATH}" \
TZ=Asia/Shanghai
WORKDIR /app
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
@@ -25,6 +26,10 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
g++ \
python3-dev \
build-essential \
tzdata \
&& ln -snf /usr/share/zoneinfo/$TZ /etc/localtime \
&& echo $TZ > /etc/timezone \
&& dpkg-reconfigure --frontend noninteractive tzdata \
&& rm -rf /var/lib/apt/lists/*
COPY common ./common
backend/pyproject.toml
+1
View File
@@ -13,6 +13,7 @@ dependencies = [
"gunicorn>=26.0.0",
"passlib==1.7.4",
"bcrypt>=4.0,<4.1",
"loguru>=0.7.2",
]
[tool.uv.sources]
backend/src/backend/main.py
+8
View File
@@ -16,6 +16,7 @@ from backend.auth import router as auth_router
from backend.jupyter import router as jupyter_router
from backend.resources import router as resources_router
from backend.runtime_client import RuntimeClient
from backend.rclone_rc_client import RcloneRCClient
from backend.schedule_runs import router as schedule_runs_router
from backend.schedules import router as schedules_router
from backend.scripts import router as scripts_router
@@ -59,9 +60,16 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
timeout=httpx.Timeout(30.0),
)
app.state.runtime_client = RuntimeClient(runtime_http_client)
# Short timeout — refresh is best-effort and runs in a BackgroundTask.
rclone_http_client = httpx.AsyncClient(
base_url=settings.rclone_rc_url,
timeout=httpx.Timeout(30.0),
)
app.state.rclone_rc_client = RcloneRCClient(rclone_http_client)
try:
yield
finally:
await rclone_http_client.aclose()
await runtime_http_client.aclose()
await storage_http_client.aclose()
await engine.dispose()
backend/src/backend/rclone_rc_client.py
+92
View File
@@ -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"]
backend/src/backend/runtime_client.py
+212 -1
View File
@@ -4,7 +4,7 @@ from dataclasses import dataclass
from typing import Any
import httpx
from loguru import logger
@dataclass(frozen=True)
class RuntimeClientError(Exception):
@@ -77,5 +77,216 @@ class RuntimeClient:
{"action": "start", "workspace_id": workspace_id},
)
async def _ensure_workspace(
self,
workspace_id: str,
) -> dict[str, Any]:
"""Return a running workspace descriptor, starting it if needed.
Mirrors the lazy-start pattern used by
:func:`backend.jupyter.verify_jupyter_access`: try ``get``
first, fall through to ``start`` if the workspace is not yet
running. Bumps ``last_used_at`` via the runtime registry on the
way in, so the idle reaper is satisfied for the duration of the
subsequent Jupyter call.
"""
ws = await self.get_workspace(workspace_id)
if not ws or ws.get("status") != "running":
ws = await self.start_workspace(workspace_id)
if not ws.get("port") or not ws.get("token"):
raise RuntimeClientError(
500,
{
"code": "JUPYTER_DESCRIPTOR_INVALID",
"message": "Runtime returned no port/token for workspace",
"retryable": False,
"details": {"workspace_id": workspace_id},
},
)
return ws
async def _jupyter_request(
self,
workspace_id: str,
ws: dict[str, Any],
method: str,
contents_path: str,
body: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Send a request to the workspace Jupyter's contents API.
``contents_path`` is the suffix after ``/api/contents/`` —
pass ``""`` for the root, ``"foo.ipynb"`` for a file, or
``"foo.ipynb/checkpoints"`` for the checkpoint subresource.
The URL is built from the per-workspace ``base_url`` + ``port``
and authenticated with the runtime-issued token. We reuse the
existing ``httpx.AsyncClient`` (its ``base_url`` is overridden
by the absolute URL we hand it).
"""
suffix = contents_path.lstrip("/")
url = (
f"{ws['base_url']}:{ws['port']}"
f"/jupyter/{workspace_id}/api/contents/{suffix}"
)
logger.debug(f"{method} {url}")
logger.debug(body)
headers = {"Authorization": f"token {ws['token']}"}
try:
if body is None:
response = await self.client.request(
method, url, headers=headers
)
else:
response = await self.client.request(
method, url, json=body, headers=headers
)
except httpx.RequestError as exc:
raise _RUNTIME_TRANSPORT_ERROR from exc
if response.is_error:
try:
detail = response.json()
except ValueError:
detail = response.text
raise RuntimeClientError(response.status_code, detail)
if response.status_code == 204 or not response.content:
return {}
return response.json()
async def create_notebook(
self,
workspace_id: str,
*,
name: str,
cells: list[dict[str, Any]] | None = None,
checkpoint: bool = True,
) -> dict[str, Any]:
"""Create a notebook at the given path in the workspace.
Uses Jupyter's ``PUT /api/contents/{name}`` (the
"save / create at path" verb — ``POST /api/contents/`` only
creates an auto-incremented ``Untitled.ipynb``). The body
carries ``type``, ``format`` and the full notebook ``content``.
Auto-starts the workspace's Jupyter if it is not yet running.
When ``checkpoint`` is true (default), follows up with a
``POST /api/contents/{name}/checkpoints`` so the file is
visible in the live notebook tree without an extra refresh —
checkpoint failures are logged but do not fail the create,
because the underlying ``PUT`` already persisted the bytes.
``name`` must be a safe basename or relative path; the Jupyter
side will reject anything that escapes the workspace root.
"""
ws = await self._ensure_workspace(workspace_id)
body = {
"type": "notebook",
"format": "json",
"content": {
"cells": cells if cells is not None else [],
"metadata": {},
"nbformat": 4,
"nbformat_minor": 5,
},
}
result = await self._jupyter_request(
workspace_id, ws, "PUT", name, body=body
)
if checkpoint:
try:
await self._jupyter_request(
workspace_id, ws, "POST", f"{name}/checkpoints"
)
except RuntimeClientError as exc:
logger.warning(
f"checkpoint after create_notebook({name}) failed: "
f"{exc.status_code} {exc.detail}"
)
return result
async def upload_file(
self,
workspace_id: str,
*,
name: str,
content: str,
content_type: str = "text/plain",
checkpoint: bool = True,
) -> dict[str, Any]:
"""Upload a text file to the workspace.
Uses ``PUT /api/contents/{name}`` with ``type="file"`` and
``format="text"``. ``content_type`` is accepted for API
symmetry with :meth:`create_notebook`; only ``text/*`` and
``application/json`` are supported here — binary uploads
require the base64 pathway which is not yet wired up and
raise :class:`ValueError`.
When ``checkpoint`` is true (default), the same follow-up
checkpoint call as :meth:`create_notebook` is made; failures
are swallowed.
"""
if not (
content_type.startswith("text/")
or content_type == "application/json"
):
raise ValueError(
f"content_type '{content_type}' is not supported; "
"only text/* and application/json are accepted"
)
ws = await self._ensure_workspace(workspace_id)
body = {
"type": "file",
"format": "text",
"content": content,
}
result = await self._jupyter_request(
workspace_id, ws, "PUT", name, body=body
)
if checkpoint:
try:
await self._jupyter_request(
workspace_id, ws, "POST", f"{name}/checkpoints"
)
except RuntimeClientError as exc:
logger.warning(
f"checkpoint after upload_file({name}) failed: "
f"{exc.status_code} {exc.detail}"
)
return result
async def delete_file(
self,
workspace_id: str,
*,
name: str,
) -> None:
"""Delete a file from the workspace Jupyter.
Uses ``DELETE /api/contents/{name}`` on the workspace Jupyter.
Jupyter returns ``204 No Content`` on success; any 4xx/5xx is
surfaced as :class:`RuntimeClientError`. Non-recursive — Jupyter
rejects directory deletes unless the directory is empty.
"""
ws = await self._ensure_workspace(workspace_id)
await self._jupyter_request(workspace_id, ws, "DELETE", name)
async def get_file(
self,
workspace_id: str,
*,
name: str,
) -> dict[str, Any]:
"""Read a file's contents descriptor from the workspace Jupyter.
Returns the Jupyter contents payload (``type``, ``content``,
``format``, ``mimetype``, ``size``, ...). For ``type="notebook"``
the ``content`` field is the notebook dict
(``cells``/``metadata``/``nbformat``); for ``type="file"`` it is
the raw text when ``format="text"`` or base64-encoded bytes when
``format="base64"``.
"""
ws = await self._ensure_workspace(workspace_id)
return await self._jupyter_request(workspace_id, ws, "GET", name)
__all__ = ["RuntimeClient", "RuntimeClientError"]
backend/src/backend/scripts.py
+301 -122
View File
@@ -7,9 +7,10 @@ import mimetypes
from datetime import UTC, datetime
from pathlib import PurePosixPath
from typing import Any
from loguru import logger
from fastapi import (
APIRouter,
BackgroundTasks,
Depends,
Header,
HTTPException,
@@ -31,6 +32,7 @@ from backend.dependencies import (
database_session,
request_context,
)
from backend.runtime_client import RuntimeClientError
from backend.schemas import (
CreateScriptRequest,
CreateWorkspaceDirectoryRequest,
@@ -98,6 +100,27 @@ def safe_script_name(value: str, script_type: str) -> str:
return name
def _jupyter_path(script_type: str, script_id: str) -> str:
"""Return the in-Jupyter path used for a script.
The path is ``{script_id}.{ext}`` — deterministic and
collision-free because ``script_id`` is a fresh ULID. Jupyter's
contents API treats the suffix after ``/api/contents/`` as a path
relative to its ``--notebook-dir`` (the workspace root), so a
``notebooks/`` prefix here would push the file into a non-existent
``notebooks/`` subdir on disk. The user's ``script_name`` is kept on
the Scripts row as a display label only; the on-disk filename is
owned by the database.
The ``/notebooks/`` segment in the user's browser URL
(``/jupyter/<ws>/notebooks/<file>.ipynb``) is jupyter's URL route
for the editor view, not a filesystem path — jupyter routes that
URL to the file at the workspace root.
"""
ext = ".ipynb" if script_type == "notebook" else ".py"
return f"{script_id}{ext}"
def validate_script_content(content: str, script_type: str) -> bytes:
encoded = content.encode("utf-8")
if len(encoded) > 10 * 1024 * 1024:
@@ -278,63 +301,85 @@ async def create_script_record(
) if parent_path is None else normalize_user_path(parent_path)
child_path = f"{folder}/{name}" if folder else name
relative_path = user_relative_path(context, child_path)
logger.debug(relative_path)
existing_script = await session.scalar(
select(Scripts)
.join(
StorageObjects,
StorageObjects.storage_object_id == Scripts.current_object_id,
)
.where(
# The Jupyter-side filename is owned by the database: a fresh ULID
# guarantees uniqueness, so the StorageObject-based 409 check from
# the legacy flow no longer applies. We still do a Scripts-only
# conflict check on the user-supplied name so two scripts cannot
# claim the same display name within the same workspace.
script_id = new_ulid()
jupyter_name = _jupyter_path(script_type, script_id)
name_clash = await session.scalar(
select(Scripts.script_id).where(
Scripts.workspace_id == context.workspace.workspace_id,
StorageObjects.relative_path == relative_path,
Scripts.script_name == name,
Scripts.status == "active",
)
)
if existing_script is not None:
if name_clash is not None:
raise HTTPException(
status.HTTP_409_CONFLICT,
"a file with the same path already exists",
"a script with the same name already exists in this workspace",
)
storage_data = await request.app.state.storage_client.create_server_object(
# Push the file directly to the workspace's live Jupyter instance.
# Auto-starts the workspace if no Jupyter is running yet. Once
# Jupyter has the file in its local mount, the rclone VFS will
# eventually replicate it back to object storage.
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
content_hash = hashlib.sha256(content).hexdigest()
size_bytes = len(content)
try:
if script_type == "notebook":
notebook = json.loads(content.decode("utf-8"))
logger.debug(notebook)
jupyter_resp = await runtime_client.create_notebook(
workspace_id,
name=jupyter_name,
cells=notebook.get("cells"),
)
else:
jupyter_resp = await runtime_client.upload_file(
workspace_id,
name=jupyter_name,
content=content.decode("utf-8"),
content_type=(
mimetypes.guess_type(jupyter_name)[0] or "text/plain"
),
)
logger.debug(jupyter_resp)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
detail=exc.detail,
) from exc
# Synthesise a storage_data-shaped dict from the Jupyter response
# so the existing script_payload + response shape keep working.
# The storage_object_id is a fresh ULID — there is no real
# StorageObject row for this file; downstream list/get operations
# that JOIN StorageObjects will skip jupyter-only scripts.
object_id = new_ulid()
storage_data = {
"storage_object_id": object_id,
"relative_path": jupyter_name,
"object_key": f"{workspace_id}/{jupyter_name}",
"content_hash": content_hash,
"size_bytes": size_bytes,
}
script = Scripts(
script_id=script_id,
workspace_id=context.workspace.workspace_id,
user_id=context.user.user_id,
usage_type="working_copy",
file_name=name,
content_type=mimetypes.guess_type(name)[0]
or "application/octet-stream",
content=content,
current_object_id=object_id,
owner_user_id=context.user.user_id,
script_name=name,
script_type=script_type,
visibility=visibility,
is_immutable=False,
idempotency_key=f"script:{context.workspace.workspace_id}:{relative_path}",
relative_path=relative_path,
status="active",
)
object_id = storage_data["storage_object_id"]
script = await session.scalar(
select(Scripts).where(Scripts.current_object_id == object_id)
)
now = datetime.now(UTC).replace(tzinfo=None)
if script is None:
script = Scripts(
script_id=new_ulid(),
workspace_id=context.workspace.workspace_id,
current_object_id=object_id,
owner_user_id=context.user.user_id,
script_name=name,
script_type=script_type,
visibility=visibility,
status="active",
)
session.add(script)
else:
script.owner_user_id = context.user.user_id
script.script_name = name
script.script_type = script_type
script.visibility = visibility
script.status = "active"
script.deleted_at = None
script.updated_at = now
session.add(script)
await session.flush()
await session.refresh(script)
return script, storage_data
@@ -344,6 +389,7 @@ async def create_script_record(
async def create_script(
payload: CreateScriptRequest,
request: Request,
background_tasks: BackgroundTasks,
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
@@ -359,6 +405,11 @@ async def create_script(
context=context,
session=session,
)
# background_tasks.add_task(
# request.app.state.rclone_rc_client.vfs_refresh,
# dir_path=context.workspace.workspace_id,
# recursive=True,
# )
return {
"request_id": context.request_id,
"data": script_payload(script, storage_data),
@@ -372,6 +423,7 @@ async def create_script(
)
async def upload_script(
request: Request,
background_tasks: BackgroundTasks,
file_name: str = Query(min_length=1, max_length=255),
parent_path: str = Query(default="", max_length=1024),
visibility: str = Query(
@@ -413,6 +465,11 @@ async def upload_script(
context=context,
session=session,
)
# background_tasks.add_task(
# request.app.state.rclone_rc_client.vfs_refresh,
# dir_path=context.workspace.workspace_id,
# recursive=True,
# )
return {
"request_id": context.request_id,
"data": script_payload(script, storage_data),
@@ -539,29 +596,45 @@ async def delete_workspace_directory(
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
directory_path = normalize_user_path(path, allow_empty=False)
relative_prefix = user_relative_path(context, directory_path)
# The original query JOINed StorageObjects to filter scripts by
# ``relative_path`` prefix. Jupyter-only scripts have no
# StorageObject row, and the Scripts schema does not track
# directory membership, so we cannot honour the directory filter
# for jupyter-created files. We therefore enumerate all active
# scripts the user owns in the workspace and forward each delete
# to the runtime. ``directory_path`` is still returned in the
# response for API compatibility.
rows = (
await session.execute(
select(Scripts, StorageObjects)
.join(
StorageObjects,
StorageObjects.storage_object_id == Scripts.current_object_id,
)
select(Scripts)
.where(
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.owner_user_id == context.user.user_id,
Scripts.status == "active",
StorageObjects.relative_path.startswith(
f"{relative_prefix}/"
),
)
)
).all()
for script, _storage in rows:
await request.app.state.storage_client.delete_object(
script.current_object_id
)
).scalars().all()
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
for script in rows:
# Best-effort: try to delete from jupyter, but do not let one
# failure abort the rest. The jupyter call will 404 if the
# file is not actually in the workspace (e.g. legacy scripts
# whose on-disk filename we do not know); we treat that as a
# no-op and still mark the row deleted.
try:
await runtime_client.delete_file(
workspace_id,
name=_jupyter_path(script.script_type, script.script_id),
)
except RuntimeClientError as exc:
if exc.status_code != 404:
logger.warning(
f"delete_workspace_directory: jupyter delete "
f"failed for {script.script_id} ({script.script_name}): "
f"{exc.status_code} {exc.detail}"
)
script.status = "deleted"
script.deleted_at = datetime.now(UTC).replace(tzinfo=None)
return {
@@ -630,48 +703,65 @@ async def update_script(
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
script, storage_object = await get_script_row(
script_id,
context,
session,
for_update=True,
# Look up the Scripts row on its own: jupyter-only scripts do not
# have a StorageObjects row to JOIN against, and update is an
# in-place overwrite of the same jupyter path, so we do not need
# any object-store metadata.
script = await session.scalar(
select(Scripts)
.where(
Scripts.script_id == script_id,
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
)
.with_for_update()
)
if script is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"script not found",
)
require_script_modify_access(
script,
user_id=context.user.user_id,
is_admin=context.is_admin,
)
content = validate_script_content(payload.content, script.script_type)
if not storage_object.relative_path:
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
jupyter_name = _jupyter_path(script.script_type, script.script_id)
try:
if script.script_type == "notebook":
notebook = json.loads(content.decode("utf-8"))
jupyter_resp = await runtime_client.create_notebook(
workspace_id,
name=jupyter_name,
cells=notebook.get("cells"),
)
else:
jupyter_resp = await runtime_client.upload_file(
workspace_id,
name=jupyter_name,
content=content.decode("utf-8"),
content_type=(
mimetypes.guess_type(jupyter_name)[0]
or "text/plain"
),
)
except RuntimeClientError as exc:
raise HTTPException(
status.HTTP_409_CONFLICT,
"script has no workspace path",
)
old_object_id = script.current_object_id
storage_data = await request.app.state.storage_client.create_server_object(
workspace_id=context.workspace.workspace_id,
user_id=script.owner_user_id,
usage_type="working_copy",
file_name=storage_object.file_name or script.script_name,
content_type=storage_object.mime_type
or mimetypes.guess_type(script.script_name)[0]
or "application/octet-stream",
content=content,
visibility=script.visibility,
is_immutable=False,
idempotency_key=(
f"script-update:{script.script_id}:"
f"{hashlib.sha256(content).hexdigest()}"
),
relative_path=storage_object.relative_path,
)
script.current_object_id = storage_data["storage_object_id"]
status_code=exc.status_code,
detail=exc.detail,
) from exc
script.updated_at = datetime.now(UTC).replace(tzinfo=None)
if old_object_id != script.current_object_id:
try:
await request.app.state.storage_client.delete_object(old_object_id)
except Exception:
pass
storage_data = {
"storage_object_id": script.current_object_id,
"relative_path": jupyter_name,
"object_key": f"{workspace_id}/{jupyter_name}",
"content_hash": hashlib.sha256(content).hexdigest(),
"size_bytes": len(content),
}
return {
"request_id": context.request_id,
"data": script_payload(script, storage_data),
@@ -686,20 +776,41 @@ async def delete_script(
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
script, storage_object = await get_script_row(
script_id,
context,
session,
for_update=True,
# Scripts created via the jupyter path do not have a corresponding
# StorageObjects row, so we look up the Scripts row on its own and
# forward the delete to the workspace's live Jupyter instance.
script = await session.scalar(
select(Scripts)
.where(
Scripts.script_id == script_id,
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
)
.with_for_update()
)
if script is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"script not found",
)
require_script_modify_access(
script,
user_id=context.user.user_id,
is_admin=context.is_admin,
)
await request.app.state.storage_client.delete_object(
script.current_object_id
)
runtime_client = request.app.state.runtime_client
try:
await runtime_client.delete_file(
context.workspace.workspace_id,
name=_jupyter_path(script.script_type, script.script_id),
)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
detail=exc.detail,
) from exc
script.status = "deleted"
script.deleted_at = datetime.now(UTC).replace(tzinfo=None)
return {
@@ -724,12 +835,25 @@ async def publish_version(
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
script, source_object = await get_script_row(
script_id,
context,
session,
for_update=True,
# Jupyter-only scripts do not have a StorageObjects row, so the
# legacy get_script_row helper raises 409 before we even get here.
# Look up the Scripts row on its own — version publication is a
# metadata operation, we do not need the working-copy object
# metadata.
script = await session.scalar(
select(Scripts)
.where(
Scripts.script_id == script_id,
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
)
.with_for_update()
)
if script is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"script not found",
)
require_script_modify_access(
script,
user_id=context.user.user_id,
@@ -743,22 +867,29 @@ async def publish_version(
status.HTTP_412_PRECONDITION_FAILED,
"source_object_id is not the current working copy",
)
if not source_object.relative_path:
raise HTTPException(
status.HTTP_409_CONFLICT,
"script has no workspace path",
)
if not source_object.bucket_name or not source_object.object_key:
raise HTTPException(
status.HTTP_409_CONFLICT,
"script working copy is not stored in object storage",
)
content = await asyncio.to_thread(
request.app.state.object_store.get_bytes,
bucket_name=source_object.bucket_name,
object_key=source_object.object_key,
)
# Read the working-copy content from the workspace's Jupyter
# instance. Notebooks come back as a dict (json.dumps it); text
# files come back as a UTF-8 string.
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
jupyter_name = _jupyter_path(script.script_type, script.script_id)
try:
contents = await runtime_client.get_file(
workspace_id, name=jupyter_name
)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
detail=exc.detail,
) from exc
if contents.get("type") == "notebook":
content = json.dumps(
contents.get("content", {}), ensure_ascii=False
).encode("utf-8")
else:
content = (contents.get("content") or "").encode("utf-8")
content_hash = hashlib.sha256(content).hexdigest()
existing = await session.scalar(
select(Versions).where(
@@ -801,7 +932,7 @@ async def publish_version(
artifact_object_id=artifact["storage_object_id"],
version_no=version_no,
version_label=f"v{version_no}.0",
source_path=source_object.relative_path,
source_path=jupyter_name,
artifact_path=artifact["storage_uri"],
content_hash=content_hash,
file_size_bytes=len(content),
@@ -840,6 +971,54 @@ async def list_versions(
}
@router.get("/api/v1/scripts/{script_id}/latest-version")
async def latest_version(
script_id: str,
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Return just the latest version's ``version_label`` + ``versions_id``.
Lightweight alternative to :func:`list_versions` for the editor
header that only needs to show "vN · <id>". Validates the script
exists in the caller's workspace, then queries the single most
recent version row. Returns ``data: null`` when the script has no
published versions yet (so the frontend can render an empty label
without a 404 round-trip).
"""
script = await session.scalar(
select(Scripts.script_id).where(
Scripts.script_id == script_id,
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
)
)
if script is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"script not found",
)
latest = await session.scalar(
select(Versions).where(
Versions.script_id == script_id,
).order_by(Versions.version_no.desc()).limit(1)
)
if latest is None:
return {
"request_id": context.request_id,
"data": None,
"meta": {"has_versions": False},
}
return {
"request_id": context.request_id,
"data": {
"versions_id": latest.versions_id,
"version_label": latest.version_label,
},
"meta": {"has_versions": True},
}
@router.get("/api/v1/versions/{versions_id}")
async def get_version(
versions_id: str,
common/src/common/config.py
+4
View File
@@ -54,6 +54,10 @@ class Settings(BaseSettings):
default="http://runtime:8000",
description="Backend → Runtime HTTP endpoint.",
)
rclone_rc_url: str = Field(
default="http://runtime:5572",
description="Backend → rclone RC HTTP endpoint (VFS cache invalidation).",
)
# ── RustFS object storage ────────────────────────────────────
rustfs_endpoint: str = Field(
common/src/common/utils.py
+8
View File
@@ -12,6 +12,7 @@ specific concerns:
from __future__ import annotations
import os
import socket
import subprocess
import time
@@ -30,6 +31,7 @@ def start_process(
cmd: list[str],
workspace_path: Path,
log_dir: str | Path = "/tmp/process_logs",
env: dict[str, str] | None = None,
) -> tuple[subprocess.Popen, Path]:
"""Launch ``cmd`` as a subprocess and return ``(process, log_file)``.
@@ -44,6 +46,11 @@ def start_process(
start_time = time.strftime("%Y%m%d_%H%M%S")
temp_log = log_dir_path / f"process_start_{start_time}.log"
# 构建合并后的环境变量
full_env = os.environ.copy()
if env:
full_env.update(env)
with open(temp_log, "a", buffering=1) as log_file:
process = subprocess.Popen(
cmd,
@@ -51,6 +58,7 @@ def start_process(
stdout=log_file,
stderr=subprocess.STDOUT,
start_new_session=True,
env=full_env,
)
final_log = log_dir_path / f"process_{process.pid}_{start_time}.log"
docker-compose.yml
+1 -3
View File
@@ -115,9 +115,7 @@ services:
volumes:
- ./runtime:/app/runtime
healthcheck:
test:
- CMD-SHELL
- grep -q ' /app/workspaces .* - fuse.rclone ' /proc/self/mountinfo && curl -fsS http://127.0.0.1:8000/api/v1/health >/dev/null
test: ["CMD-SHELL", "grep -q ' /app/workspaces .* - fuse.rclone ' /proc/self/mountinfo && curl -fsS http://127.0.0.1:8000/api/v1/health >/dev/null"]
interval: 10s
timeout: 5s
retries: 18
frontend/Dockerfile
+4
View File
@@ -7,6 +7,10 @@ COPY frontend/ ./
RUN pnpm typecheck && pnpm build
FROM nginx:alpine
ENV TZ=Asia/Shanghai
RUN apk add --no-cache tzdata \
&& ln -snf /usr/share/zoneinfo/$TZ /etc/localtime \
&& echo $TZ > /etc/timezone
COPY ./default.conf /etc/nginx/conf.d/default.conf.template
COPY ./scripts/nginx-entrypoint.sh /usr/local/bin/model-platform-entrypoint.sh
RUN sed -i 's/\r$//' /usr/local/bin/model-platform-entrypoint.sh \
frontend/app/context/AuthContext.tsx
+2
View File
@@ -220,6 +220,8 @@ export function useApi(): WorkspaceBoundApi {
rawApi.releaseFileLockOnUnload(workspaceId, session),
createJupyterAccessTicket: (session) =>
rawApi.createJupyterAccessTicket(workspaceId, session),
getLatestScriptVersion: (scriptId) =>
rawApi.getLatestScriptVersion(workspaceId, scriptId),
listScriptVersions: (scriptId) => rawApi.listScriptVersions(workspaceId, scriptId),
publishScriptVersion: (input) => rawApi.publishScriptVersion(workspaceId, input),
listSchedules: () => rawApi.listSchedules(workspaceId),
frontend/app/features/platform/ModelPlatformApp.tsx
+27 -2
View File
@@ -11,6 +11,7 @@ import { useLocation, useNavigate } from "react-router";
import {
type ActiveEditSession,
type LatestVersion,
type ScriptItem,
type ScriptType,
type StableVersion,
@@ -163,6 +164,10 @@ function AuthenticatedModelPlatformApp() {
} | null>(null);
const [versions, setVersions] = useState<StableVersion[]>([]);
const [versionsLoading, setVersionsLoading] = useState(false);
const [latestVersion, setLatestVersion] = useState<LatestVersion | null>(
null,
);
const [latestVersionLoading, setLatestVersionLoading] = useState(false);
const [publishTarget, setPublishTarget] = useState<ScriptItem | null>(null);
const [releaseNote, setReleaseNote] = useState("");
const [publishVisibility, setPublishVisibility] =
@@ -244,10 +249,30 @@ function AuthenticatedModelPlatformApp() {
useEffect(() => {
if (!selectedId) {
setVersions([]);
setLatestVersion(null);
return;
}
let ignore = false;
setVersionsLoading(true);
setLatestVersionLoading(true);
void api.getLatestScriptVersion(selectedId)
.then((item) => {
if (!ignore) setLatestVersion(item);
})
.catch((error) => {
if (!ignore) {
setToast({
tone: "error",
message:
error instanceof Error
? error.message
: "最新版本加载失败",
});
}
})
.finally(() => {
if (!ignore) setLatestVersionLoading(false);
});
void api.listScriptVersions(selectedId)
.then((items) => {
if (!ignore) setVersions(items);
@@ -1007,8 +1032,8 @@ function AuthenticatedModelPlatformApp() {
? editorOpenError.message
: null
}
latestVersion={versions[0] ?? null}
versionsLoading={versionsLoading}
latestVersion={latestVersion}
versionsLoading={latestVersionLoading}
onOpenEditor={() => void openScriptEditor(selected)}
onEndEditing={() => void endEditing()}
onClose={() => {
frontend/app/features/platform/ScriptWorkspace.tsx
+2 -2
View File
@@ -1,8 +1,8 @@
import Icon from "../../components/Icon";
import type {
ActiveEditSession,
LatestVersion,
ScriptItem,
StableVersion,
} from "../../services/api";
import { scriptIcon } from "./WorkspaceTree";
@@ -17,7 +17,7 @@ type ScriptWorkspaceProps = {
jupyterUrl: string | null;
editBusy: boolean;
openError: string | null;
latestVersion: StableVersion | null;
latestVersion: LatestVersion | null;
versionsLoading: boolean;
onOpenEditor: () => void;
onEndEditing: () => void;
frontend/app/services/api.ts
+17
View File
@@ -558,6 +558,22 @@ export async function createJupyterAccessTicket(
};
}
export type LatestVersion = {
versions_id: string;
version_label: string;
};
export async function getLatestScriptVersion(
workspaceId: string,
scriptId: string,
): Promise<LatestVersion | null> {
return apiRequest<LatestVersion | null>(
`/api/v1/scripts/${scriptId}/latest-version`,
{},
workspaceId,
);
}
export async function listScriptVersions(
workspaceId: string,
scriptId: string,
@@ -1102,6 +1118,7 @@ export type WorkspaceBoundApi = {
createJupyterAccessTicket: (
session: ActiveEditSession,
) => Promise<JupyterAccessTicket>;
getLatestScriptVersion: (scriptId: string) => Promise<LatestVersion | null>;
listScriptVersions: (scriptId: string) => Promise<StableVersion[]>;
publishScriptVersion: (
input: Parameters<typeof publishScriptVersion>[1],
runtime/Dockerfile
+10 -1
View File
@@ -3,7 +3,8 @@ FROM python:3.12-slim-bookworm
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PYTHONPATH=/app \
PATH="/app/.venv/bin:${PATH}"
PATH="/app/.venv/bin:${PATH}" \
TZ=Asia/Shanghai
WORKDIR /app
# 从官方 Rclone 镜像直接复制 rclone 二进制文件（高效、稳定）
COPY --from=rclone/rclone:latest /usr/local/bin/rclone /usr/local/bin/rclone
@@ -28,6 +29,10 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
procps \
build-essential \
python3-dev \
tzdata \
&& ln -snf /usr/share/zoneinfo/$TZ /etc/localtime \
&& echo $TZ > /etc/timezone \
&& dpkg-reconfigure --frontend noninteractive tzdata \
&& sed -i 's/#user_allow_other/user_allow_other/g' /etc/fuse.conf \
&& rm -rf /var/lib/apt/lists/*
@@ -37,5 +42,9 @@ COPY common ./common
COPY runtime ./runtime
RUN UV_DEFAULT_INDEX="https://pypi.tuna.tsinghua.edu.cn/simple/" uv sync --frozen --no-dev --no-editable --package runtime
RUN mkdir -p /root/.jupyter/custom /app/.venv/share/jupyter/custom
RUN echo '#top-panel { display: none !important; height: 0 !important; }' > /root/.jupyter/custom/custom.css && \
cp /root/.jupyter/custom/custom.css /app/.venv/share/jupyter/custom/custom.css
EXPOSE 8000
CMD ["uv", "run", "--frozen", "--package", "runtime", "gunicorn", "--config", "runtime/gunicorn.conf.py", "runtime.main:app"]
runtime/src/runtime/mount.py
+3
View File
@@ -68,6 +68,9 @@ def start_rclone_mount() -> None:
"--dir-cache-time", "30s",
"--poll-interval", "30s",
"--log-level", "INFO",
"--rc",
"--rc-addr", "0.0.0.0:5572",
"--rc-no-auth",
]
RCLONE_PROCESS = subprocess.Popen(
cmd,
runtime/src/runtime/process.py
+15 -10
View File
@@ -160,7 +160,7 @@ def _bump_last_used(record: JupyterProcessRecord) -> None:
async def start_workspace(ws_id: str) -> dict:
async with get_workspace_lock(ws_id):
workspace_path = WORKSPACES_ROOT / ws_id
workspace_path = (WORKSPACES_ROOT / ws_id).resolve()
workspace_path.mkdir(parents=True, exist_ok=True)
if ws_id in JUPYTER_PROCESSES:
@@ -195,14 +195,16 @@ async def start_workspace(ws_id: str) -> dict:
base_path = f"/jupyter/{ws_id}/"
cmd = [
"jupyter", "notebook",
"jupyter",
"notebook",
f"--port={port}",
"--ip=0.0.0.0",
"--no-browser",
"--allow-root",
f"--ServerApp.token={token}",
f"--ServerApp.base_url={base_path}",
"--notebook-dir=.",
# f"--notebook-dir={workspace_path}",
f"--ServerApp.root_dir={workspace_path}",
"--ServerApp.terminals_enabled=False",
"--NotebookApp.terminals_enabled=False",
"--ServerApp.allow_origin=*",
@@ -212,12 +214,14 @@ async def start_workspace(ws_id: str) -> dict:
]
try:
process, log_file = start_process(cmd, workspace_path)
process, log_file = start_process(
cmd,
WORKSPACES_ROOT,
env={"PATH": f"/app/.venv/bin:{os.environ.get('PATH', '')}"},
)
except Exception as e:
logger.error(f"Failed to start Jupyter for {ws_id}: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to start Jupyter: {e}"
)
raise HTTPException(status_code=500, detail=f"Failed to start Jupyter: {e}")
full_url = f"{PUBLIC_BASE_URL}:{port}{base_path}?token={token}"
meta_path = _meta_path(ws_id)
@@ -341,8 +345,7 @@ async def get_workspace(ws_id: str) -> dict:
raise HTTPException(
status_code=404,
detail=(
f"Jupyter process for workspace '{ws_id}' "
"has terminated unexpectedly."
f"Jupyter process for workspace '{ws_id}' " "has terminated unexpectedly."
),
)
@@ -416,7 +419,9 @@ def reconcile_processes() -> dict[str, int]:
except PermissionError:
alive = True # someone else's process, leave alone
if not alive:
logger.info(f"reconcile: dropping stale sidecar for {entry} (pid {pid} dead)")
logger.info(
f"reconcile: dropping stale sidecar for {entry} (pid {pid} dead)"
)
_delete_meta(entry)
counters["removed_meta"] += 1
return counters
schedule/Dockerfile
+6 -1
View File
@@ -3,7 +3,8 @@ FROM python:3.12-slim-bookworm
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PYTHONPATH=/app \
PATH="/app/.venv/bin:${PATH}"
PATH="/app/.venv/bin:${PATH}" \
TZ=Asia/Shanghai
WORKDIR /app
RUN ( \
@@ -24,6 +25,10 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
g++ \
python3-dev \
build-essential \
tzdata \
&& ln -snf /usr/share/zoneinfo/$TZ /etc/localtime \
&& echo $TZ > /etc/timezone \
&& dpkg-reconfigure --frontend noninteractive tzdata \
&& rm -rf /var/lib/apt/lists/*
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
schedule/pyproject.toml
+1
View File
@@ -13,6 +13,7 @@ dependencies = [
"nbformat==5.10.4",
"ipykernel==6.29.5",
"gunicorn>=26.0.0",
"loguru>=0.7.2",
]
[tool.uv.sources]
schedule/src/schedule/orchestrator.py
+4 -7
View File
@@ -13,10 +13,10 @@ post-back (see ``schedule.service.trigger_schedule``).
from __future__ import annotations
import asyncio
import logging
from datetime import timedelta
from typing import Any
from loguru import logger
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
@@ -36,13 +36,10 @@ from schedule.context import (
TERMINAL_RUN_STATES,
)
LOGGER = logging.getLogger(__name__)
SCHEDULE_RUN_REQUESTED_EVENT = schedule_event_type("schedule.run.requested")
NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute")
NODE_FINISHED_EVENT = schedule_event_type("job.node.finished")
class DispatchOrchestrator:
"""Polls Outbox + advances DAG schedule runs.
@@ -133,7 +130,7 @@ class DispatchOrchestrator:
except asyncio.CancelledError:
raise
except Exception:
LOGGER.exception("database event loop failed")
logger.exception("database event loop failed")
await asyncio.sleep(1)
async def _execution_loop(self) -> None:
@@ -145,7 +142,7 @@ class DispatchOrchestrator:
except asyncio.CancelledError:
raise
except Exception:
LOGGER.exception("node execute loop failed")
logger.exception("node execute loop failed")
await asyncio.sleep(1)
async def _claim_execution_events(
@@ -233,7 +230,7 @@ class DispatchOrchestrator:
.with_for_update()
)
if item is None:
LOGGER.warning("execution event %s disappeared", event_id)
logger.warning("execution event {} disappeared", event_id)
return
if exc is None:
item.event_status = "published"
schedule/src/schedule/scheduler.py
+3 -4
View File
@@ -15,11 +15,12 @@ restarts.
from __future__ import annotations
import asyncio
import logging
from datetime import UTC
from typing import Awaitable, Callable
from zoneinfo import ZoneInfo
from loguru import logger
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from sqlalchemy import select
@@ -31,8 +32,6 @@ from common.scheduler import build_sqlalchemy_jobstore
from schedule.context import naive_utc
LOGGER = logging.getLogger(__name__)
# APScheduler's persistent SQLAlchemy job store pickles each job. A bound
# ``CronScheduler`` method captures this instance (including SQLAlchemy engine
# state) and therefore cannot be pickled. Keep the persisted callable at
@@ -111,7 +110,7 @@ class CronScheduler:
except asyncio.CancelledError:
raise
except Exception:
LOGGER.exception("cron job synchronization failed")
logger.exception("cron job synchronization failed")
await asyncio.sleep(5)
async def _sync_once(self) -> None:
schedule/src/schedule/service.py
+3 -5
View File
@@ -20,11 +20,11 @@ service shares the same MySQL via the Docker network).
from __future__ import annotations
import logging
from datetime import UTC, datetime
from typing import Any
import httpx
from loguru import logger
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from common.config import settings
@@ -39,8 +39,6 @@ from schedule.orchestrator import DispatchOrchestrator
from schedule.scheduler import CronScheduler
from schedule.worker import NodeExecutor
LOGGER = logging.getLogger(__name__)
class SchedulerService:
"""Composes cron / orchestrator / worker into one bootable service.
@@ -153,8 +151,8 @@ class SchedulerService:
except TriggerError as exc:
# Most likely: idempotency_key collision from a previous
# tick in the same minute — silently no-op.
LOGGER.info(
"cron trigger no-op for schedule %s: %s", schedule_id, exc,
logger.info(
"cron trigger no-op for schedule {}: {}", schedule_id, exc,
)
async def process_pending_events(
schedule/src/schedule/worker.py
+2 -5
View File
@@ -16,11 +16,11 @@ from __future__ import annotations
import asyncio
import hashlib
import json
import logging
import traceback
from pathlib import Path
from typing import Any
from loguru import logger
from sqlalchemy import select
from common.db import session_scope
@@ -43,12 +43,9 @@ from common.eventing import (
from schedule.context import TERMINAL_NODE_STATES
from schedule.execution import ExecutionResult, execute_artifact
LOGGER = logging.getLogger(__name__)
NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute")
NODE_FINISHED_EVENT = schedule_event_type("job.node.finished")
class NodeExecutor:
"""Owns the actual execution of one schedule node (notebook / python)."""
@@ -376,7 +373,7 @@ class NodeExecutor:
result_id = result_object["storage_object_id"]
except Exception as exc:
upload_error = f"result upload failed: {exc}"[:2000]
LOGGER.exception("failed to upload node execution artifacts")
logger.exception("failed to upload node execution artifacts")
return log_id, result_id, upload_error
@staticmethod
uv.lock
Generated
+4
View File
@@ -201,6 +201,7 @@ dependencies = [
{ name = "fastapi" },
{ name = "gunicorn" },
{ name = "httpx" },
{ name = "loguru" },
{ name = "passlib" },
{ name = "uvicorn", extra = ["standard"] },
]
@@ -215,6 +216,7 @@ requires-dist = [
{ name = "fastapi", specifier = "==0.116.1" },
{ name = "gunicorn", specifier = ">=26.0.0" },
{ name = "httpx", specifier = "==0.28.1" },
{ name = "loguru", specifier = ">=0.7.2" },
{ name = "passlib", specifier = "==1.7.4" },
{ name = "uvicorn", extras = ["standard"], specifier = "==0.35.0" },
]
@@ -1955,6 +1957,7 @@ dependencies = [
{ name = "gunicorn" },
{ name = "httpx" },
{ name = "ipykernel" },
{ name = "loguru" },
{ name = "nbclient" },
{ name = "nbformat" },
{ name = "pymysql" },
@@ -1969,6 +1972,7 @@ requires-dist = [
{ name = "gunicorn", specifier = ">=26.0.0" },
{ name = "httpx", specifier = "==0.28.1" },
{ name = "ipykernel", specifier = "==6.29.5" },
{ name = "loguru", specifier = ">=0.7.2" },
{ name = "nbclient", specifier = "==0.10.2" },
{ name = "nbformat", specifier = "==5.10.4" },
{ name = "pymysql", specifier = "==1.2.0" },