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

fix: stabilize cron scheduling and run history

Browse Source
This commit is contained in:
Winnie
2026-08-17 18:55:31 +08:00
parent 23823325e6
commit ff700d8db4
16 changed files with 320 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/backend/admin.py
+11
View File
@@ -1,3 +1,9 @@
"""旧版工作区级员工管理接口。
路由前缀为 ``/api/v1/admin``，依赖当前工作区的管理员权限。新的系统级用户、
工作区和成员管理接口在 ``platform.py``；本模块主要保留给兼容旧前端调用。
"""
from __future__ import annotations
from typing import Any, Literal
@@ -39,6 +45,7 @@ class EmployeeUpdate(BaseModel):
def require_admin(context: RequestContext) -> None:
# 统一在路由入口处做角色判断，避免每个 CRUD 分支重复写权限代码。
if not context.is_admin:
raise HTTPException(status.HTTP_403_FORBIDDEN, "仅管理员可以管理员工")
@@ -80,6 +87,7 @@ async def member_row(
return row
# 旧版接口：列出当前工作区内的员工及其角色。
@router.get("/employees")
async def list_employees(
context: RequestContext = Depends(request_context),
@@ -107,6 +115,7 @@ async def list_employees(
}
# 旧版接口：在当前工作区中创建员工及成员关系。
@router.post("/employees", status_code=status.HTTP_201_CREATED)
async def create_employee(
payload: EmployeeCreate,
@@ -160,6 +169,7 @@ async def create_employee(
}
# 旧版接口：更新员工显示信息、角色或状态。
@router.patch("/employees/{user_id}")
async def update_employee(
user_id: str,
@@ -215,6 +225,7 @@ async def update_employee(
}
# 旧版接口：移除当前工作区中的员工成员关系。
@router.delete("/employees/{user_id}")
async def delete_employee(
user_id: str,
backend/src/backend/auth.py
+16 -13
View File
@@ -1,4 +1,10 @@
"""Cookie+JWT authentication endpoints.
"""登录与会话认证接口。
中文导读：浏览器调用登录接口后，后端把 JWT 放入 HttpOnly Cookie；之后前端
通过 ``fetch(..., credentials='same-origin')`` 自动携带 Cookie。需要工作区的
接口会继续由 ``request_context`` 校验 ``workspace_id`` 和成员权限。
Cookie+JWT authentication endpoints.
The user-facing flow is:
1. POST /api/v1/auth/login — verify password, set HttpOnly cookie
@@ -30,10 +36,8 @@ from backend.dependencies import database_session, load_user_permissions
router = APIRouter(tags=["auth"])
# Cookie config. ``secure=True`` requires HTTPS — the only safe
# assumption in production. Dev environments running on plain HTTP
# should reverse-proxy with TLS termination or set the env knob
# (future extension).
# Cookie 配置：生产环境走 HTTPS 时应设置 Secure；本地 HTTP 开发环境会根据
# 实际请求协议决定是否设置，避免浏览器因 Secure Cookie 而丢弃登录状态。
COOKIE_NAME = "access_token"
COOKIE_TTL_SECONDS = 24 * 60 * 60
COOKIE_SAMESITE = "lax"
@@ -41,12 +45,8 @@ COOKIE_SAMESITE = "lax"
def _set_session_cookie(request: Request, response: Response, token: str) -> None:
forwarded_scheme = request.headers.get("x-forwarded-proto", request.url.scheme)
# Operators running behind a TLS-terminating proxy that strips
# X-Forwarded-Proto can opt into forcing the Secure flag via
# ``settings.cookie_force_secure`` — without that override a plain
# HTTP request (no forwarded scheme, no TLS upgrade visible to the
# app) would yield an insecure cookie and modern browsers would
# silently drop it on the HTTPS round trip.
# 反向代理通常用 X-Forwarded-Proto 告诉后端原始协议；若代理未传该头，
# 可通过配置强制启用 Secure，防止 HTTPS 场景下出现不安全 Cookie。
secure = forwarded_scheme == "https" or settings.cookie_force_secure
response.set_cookie(
key=COOKIE_NAME,
@@ -95,6 +95,7 @@ def _workspace_payload(
}
# 校验账号密码，设置登录 Cookie，并返回用户可进入的工作区列表。
@router.post("/api/v1/auth/login")
async def login(
request: Request,
@@ -131,8 +132,8 @@ async def login(
"invalid username or password",
)
# Pull all active memberships. The earliest join wins as default
# because there is no `is_default` column on `workspace_members`.
# 返回用户可进入的工作区列表；当前没有默认工作区字段，因此最早加入的
# 工作区作为前端的初始选择。
rows = (
await session.execute(
select(Workspaces, Roles, WorkspaceMembers.joined_at)
@@ -196,6 +197,7 @@ async def login(
}
# 清除浏览器 Cookie，使当前会话立即失效。
@router.post("/api/v1/auth/logout")
async def logout(response: Response) -> dict[str, Any]:
"""Clear the session cookie. Idempotent."""
@@ -207,6 +209,7 @@ async def logout(response: Response) -> dict[str, Any]:
}
# 返回当前登录用户、权限和可访问工作区，用于前端初始化登录态。
@router.get("/api/v1/auth/me")
async def me(
request: Request,
backend/src/backend/dependencies.py
+22 -1
View File
@@ -1,4 +1,16 @@
"""FastAPI dependencies for the public API.
"""公共 FastAPI 依赖。
中文导读：
* ``database_session``：为一次请求提供数据库事务；成功提交、异常回滚。
* ``current_user``：只验证登录 Cookie 并得到当前用户，不关心工作区。
* ``request_context``：普通业务接口最常使用的依赖，同时验证用户、
``workspace_id``、工作区成员关系和角色权限。
路由函数把这些函数写进 ``Depends(...)`` 后，FastAPI 会先完成校验，再把
结果作为参数传给路由函数；因此业务代码无需重复解析 Cookie 或查询成员关系。
FastAPI dependencies for the public API.
Two distinct concerns live here:
@@ -48,6 +60,12 @@ ACCESS_TOKEN_COOKIE = "access_token"
@dataclass(frozen=True)
class RequestContext:
"""已完成认证和工作区授权后的请求上下文。
路由依赖 ``request_context`` 后会得到该对象，用其中的用户、工作区和角色
执行业务权限判断，避免在每个接口中重复查询。
"""
request_id: str
user: Users
workspace: Workspaces
@@ -67,6 +85,7 @@ class RequestContext:
async def database_session(request: Request) -> AsyncIterator[AsyncSession]:
# 一个请求对应一个事务范围，避免不同请求意外共用同一个 Session。
async with session_scope(request.app.state.session_factory) as session:
yield session
@@ -82,6 +101,7 @@ async def current_user(
a workspace-scoped context, or use ``Depends(current_user)`` for
workspace-agnostic endpoints (e.g. ``/api/v1/auth/me``).
"""
# 登录接口写入 HttpOnly Cookie；后续浏览器请求会自动携带它。
token = request.cookies.get(ACCESS_TOKEN_COOKIE)
if not token:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not authenticated")
@@ -122,6 +142,7 @@ async def request_context(
because they own the platform. Non-admin users still need an
active ``WorkspaceMembers`` row in an active ``Workspaces`` row.
"""
# 大多数工作区接口通过这个依赖统一完成“登录 + 工作区 + 角色”三层校验。
user = await current_user(request, session)
is_system_admin = await resolve_is_system_admin(session, user)
backend/src/backend/jupyter.py
+13
View File
@@ -1,3 +1,12 @@
"""Jupyter 访问的鉴权桥接。
浏览器访问 ``/jupyter/{workspace_id}/...`` 时，Nginx 会先向本模块的
``/api/v1/auth/jupyter`` 发起内部 auth_request。后端验证用户、工作区和文件
锁，再把实际 Jupyter 地址与内部令牌写进响应头，由 Nginx 转发请求。
本模块只负责鉴权和路由选择；真正启动、管理 Jupyter 进程的是 ``runtime`` 服务。
"""
import re
from common.auth.jwt import JwtError, verify_jwt_token
@@ -44,6 +53,8 @@ async def check_notebook_is_locked(
different user, and is currently locked. The owner is always let
through; a missing row is treated as "not owned yet" and allowed.
"""
# 锁信息保存在数据库的 Scripts 记录中，而非前端内存；因此多浏览器/多用户
# 访问同一 notebook 时也能得到一致结果。
statement = select(Scripts.owner_user_id, Scripts.is_locked).where(
Scripts.workspace_id == workspace_id,
Scripts.script_name == notebook_path,
@@ -78,6 +89,8 @@ async def load_active_membership_or_403(
) from exc
# 供 Nginx auth_request 调用：验证访问 Jupyter 的身份、成员关系和文件锁，
# 再返回应转发到的 Jupyter 地址及内部令牌。
@router.get("/api/v1/auth/jupyter")
async def verify_jupyter_access(
request: Request,
backend/src/backend/main.py
+21 -8
View File
@@ -1,3 +1,14 @@
"""后端服务总入口。
这里负责两件事：
1. 在应用启动时准备数据库、对象存储以及 Runtime 的 HTTP 客户端；
2. 将各业务模块的路由注册到同一个 FastAPI 应用中。
浏览器请求先经过 Nginx 的 ``/api/`` 反向代理，随后才会到达本文件创建的
应用。具体接口实现分别在 ``auth.py``、``scripts.py``、``schedules.py`` 等
模块中。
"""
from __future__ import annotations
import time
@@ -38,15 +49,14 @@ configure_logging(settings.log_level)
@asynccontextmanager
async def lifespan(app: Any) -> AsyncIterator[None]:
# 生命周期内创建的对象挂在 app.state 上，路由通过 Depends 或 Request
# 取得它们；这样每个请求不会重复创建数据库连接或 HTTP 客户端。
engine = create_database_engine(settings.database_url)
app.state.session_factory = create_session_factory(engine)
# Storage API is part of the backend process. Platform routers call
# the helpers in ``backend.services.storage`` directly (in-process),
# so no HTTP client is needed. The dict is keyed by the actual
# bucket name (e.g. "versions"), matching ``UploadSessions.bucket_name``
# and ``StorageObjects.bucket_name`` so call sites can do
# ``object_stores[upload.bucket_name].put(...)`` directly.
# 存储接口与业务路由运行在同一个 backend 进程中，因此直接复用存储对象，
# 不需要再通过 HTTP 调用自己。字典键使用真实桶名，便于上传会话和存储
# 对象记录直接定位对应的存储后端。
app.state.object_stores: dict[str, StorageBackend | AsyncStorageBackend] = { # noqa: F821
actual_bucket_name(purpose): create_storage(build_storage_config(purpose))
for purpose in PURPOSE_BUCKETS
@@ -63,7 +73,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
},
)
app.state.runtime_client = RuntimeClient(runtime_http_client)
# Short timeout — refresh is best-effort and runs in a BackgroundTask.
# rclone 目录缓存刷新属于尽力而为的后台动作，不能拖慢用户保存文件的请求。
rclone_http_client = httpx.AsyncClient(
base_url=settings.rclone_rc_url,
timeout=httpx.Timeout(30.0),
@@ -81,6 +91,7 @@ app = create_service_app(
settings.service_name,
lifespan=lifespan,
)
# 面向浏览器的公开 API：认证、脚本、资源、调度和系统管理。
app.include_router(auth_router)
app.include_router(jupyter_router)
app.include_router(resources_router)
@@ -90,12 +101,14 @@ app.include_router(scripts_router)
app.include_router(admin_router)
app.include_router(platform_router)
# Reuse the proven storage endpoints without running another FastAPI service.
# 内部存储接口额外加上 /internal 前缀，供后端服务间调用，不作为普通前端 API。
app.include_router(storage_api_router, prefix="/internal")
@app.middleware("http")
async def access_log(request: Request, call_next):
# 每个 HTTP 请求都记录方法、路径、状态码和耗时；排查页面请求失败时，
# Docker Desktop 中 backend 容器的 Logs 就会显示这里生成的日志。
start = time.perf_counter()
try:
response = await call_next(request)
backend/src/backend/platform.py
+34 -2
View File
@@ -1,4 +1,10 @@
"""System-admin (platform-scope) endpoints for workspace & membership management.
"""系统级管理接口。
中文导读：本模块管理全平台的用户、工作区、成员关系和角色权限。它使用
``system_admin_context`` 进行平台管理员校验，因此不要求请求者先加入某个具体
工作区；普通工作区内的业务接口则使用 ``request_context``。
System-admin (platform-scope) endpoints for workspace & membership management.
All routes under ``/api/v1/platform/*`` are gated by
:func:`system_admin_context`, which requires the requester to hold a
@@ -102,6 +108,7 @@ MEMBER_STATUS_VALUES = ("active", "disabled", "locked")
# ---------------------------------------------------------------------------
# 创建工作区时前端提交的请求体；禁止未声明字段。
class WorkspaceCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
@@ -111,6 +118,7 @@ class WorkspaceCreate(BaseModel):
description: str | None = Field(default=None, max_length=1000)
# 编辑工作区时允许修改的字段；禁用操作必须走删除接口而不是直接传状态。
class WorkspaceUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
@@ -121,6 +129,7 @@ class WorkspaceUpdate(BaseModel):
status: Literal["active", "archived"] | None = None
# 将已存在用户加入工作区的请求体；角色继承用户的平台角色。
class MemberCreate(BaseModel):
"""Add a user to a workspace. Role is inherited from the user's
platform role (Users.platform_role_id) — not set here."""
@@ -130,6 +139,7 @@ class MemberCreate(BaseModel):
user_id: str = Field(min_length=26, max_length=26)
# 更新成员在该工作区中的可用状态，不直接在这里修改平台角色。
class MemberUpdate(BaseModel):
"""Update a workspace membership's status. Role cannot be changed
via this endpoint — workspace role is always inherited from the
@@ -141,6 +151,7 @@ class MemberUpdate(BaseModel):
member_status: Literal["active", "disabled", "locked"] | None = None
# 新建平台用户的请求体；创建用户不等同于把用户加入某个工作区。
class PlatformEmployeeCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
@@ -151,6 +162,7 @@ class PlatformEmployeeCreate(BaseModel):
role_code: Literal["admin", "developer"] | None = None
# 修改平台用户资料、状态或平台角色的请求体。
class PlatformEmployeeUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
@@ -160,6 +172,7 @@ class PlatformEmployeeUpdate(BaseModel):
role_code: Literal["admin", "developer"] | None = None
# 用完整权限集合替换某个平台角色菜单权限的请求体。
class RolePermissionsPatch(BaseModel):
"""Replace a platform role's permission set wholesale.
@@ -181,6 +194,8 @@ class RolePermissionsPatch(BaseModel):
@dataclass(frozen=True)
class SystemAdminContext:
"""通过系统管理员校验后的上下文，只包含当前用户和请求追踪 ID。"""
"""Resolved identity for a system-admin request.
Carries the request id, the authenticated user row, and the resolved
@@ -197,6 +212,7 @@ async def system_admin_context(
request: Request,
session: AsyncSession = Depends(database_session),
) -> SystemAdminContext:
"""验证当前用户是否为平台管理员，供 /api/v1/platform 下的路由依赖。"""
"""Resolve the requester as a system admin.
Steps:
@@ -368,6 +384,7 @@ def _envelope(request_id: str, data: Any, meta: dict[str, Any] | None = None) ->
# ---------------------------------------------------------------------------
# 列出整个平台的非删除用户；不局限于某一个工作区。
@router.get("/employees")
async def list_platform_employees(
context: SystemAdminContext = Depends(system_admin_context),
@@ -389,6 +406,7 @@ async def list_platform_employees(
)
# 创建平台用户；后续可再通过成员接口把该用户加入工作区。
@router.post("/employees", status_code=status.HTTP_201_CREATED)
async def create_platform_employee(
payload: PlatformEmployeeCreate,
@@ -429,6 +447,7 @@ async def create_platform_employee(
)
# 更新平台用户资料、账号状态或平台角色，同时保护最少管理员等约束。
@router.patch("/employees/{user_id}")
async def update_platform_employee(
user_id: str,
@@ -577,6 +596,7 @@ async def update_platform_employee(
)
# 软删除平台用户，并级联标记其工作区成员关系为删除。
@router.delete("/employees/{user_id}")
async def delete_platform_employee(
user_id: str,
@@ -629,6 +649,7 @@ async def delete_platform_employee(
# ---------------------------------------------------------------------------
# 列出平台中全部未删除工作区。
@router.get("/workspaces")
async def list_workspaces(
context: SystemAdminContext = Depends(system_admin_context),
@@ -656,6 +677,7 @@ async def list_workspaces(
)
# 创建工作区，并将当前系统管理员初始化为该工作区管理员。
@router.post("/workspaces", status_code=status.HTTP_201_CREATED)
async def create_workspace(
payload: WorkspaceCreate,
@@ -702,6 +724,7 @@ async def create_workspace(
return _envelope(context.request_id, workspace_payload(workspace))
# 读取单个工作区详情，包含已归档或禁用状态。
@router.get("/workspaces/{workspace_id}")
async def get_workspace(
workspace_id: str,
@@ -713,6 +736,7 @@ async def get_workspace(
return _envelope(context.request_id, workspace_payload(workspace))
# 更新工作区可编辑属性，例如名称、配额、描述和归档状态。
@router.patch("/workspaces/{workspace_id}")
async def update_workspace(
workspace_id: str,
@@ -740,6 +764,7 @@ async def update_workspace(
return _envelope(context.request_id, workspace_payload(workspace))
# 软删除/禁用工作区，并级联处理其活动成员关系。
@router.delete("/workspaces/{workspace_id}")
async def delete_workspace(
workspace_id: str,
@@ -780,6 +805,7 @@ async def delete_workspace(
# ---------------------------------------------------------------------------
# 列出一个工作区的活动成员与成员状态。
@router.get("/workspaces/{workspace_id}/members")
async def list_members(
workspace_id: str,
@@ -811,6 +837,7 @@ async def list_members(
)
# 将已有平台用户加入指定工作区。
@router.post(
"/workspaces/{workspace_id}/members",
status_code=status.HTTP_201_CREATED,
@@ -894,6 +921,7 @@ async def add_member(
return _envelope(context.request_id, member_payload(user, role, membership))
# 更新成员状态，例如禁用或锁定；同时保证工作区不会失去最后一个管理员。
@router.patch("/workspaces/{workspace_id}/members/{user_id}")
async def update_member(
workspace_id: str,
@@ -952,6 +980,7 @@ async def update_member(
return _envelope(context.request_id, member_payload(user, role, membership))
# 移除某个工作区成员，并保护最后一名管理员及当前操作者的安全约束。
@router.delete("/workspaces/{workspace_id}/members/{user_id}")
async def remove_member(
workspace_id: str,
@@ -1062,6 +1091,7 @@ def _role_payload(role: Roles, permission_codes: list[str]) -> dict[str, Any]:
}
# 列出平台角色及其拥有的菜单权限代码。
@router.get("/roles")
async def list_platform_roles(
context: SystemAdminContext = Depends(system_admin_context),
@@ -1086,6 +1116,7 @@ async def list_platform_roles(
)
# 获取一个角色当前配置的权限代码集合。
@router.get("/roles/{role_code}/permissions")
async def get_role_permissions(
role_code: str,
@@ -1100,6 +1131,7 @@ async def get_role_permissions(
)
# 以请求中的完整集合更新角色权限，并保留管理员角色的必要系统权限。
@router.patch("/roles/{role_code}/permissions")
async def patch_role_permissions(
role_code: str,
@@ -1242,4 +1274,4 @@ __all__ = [
"SystemAdminContext",
"router",
"system_admin_context",
]
]
backend/src/backend/resources.py
+15
View File
@@ -1,3 +1,10 @@
"""工作区数据资源 API。
资源上传分为三步：创建上传会话 → 写入文件字节 → 绑定为可见的数据资源。
这种拆分使前端可以分别处理元数据、文件传输和最终展示；资源文件本身由存储层
保存，数据库只保存资源与存储对象的关联关系。
"""
from __future__ import annotations
import os
@@ -134,6 +141,7 @@ def can_view(resource: DataResources, context: RequestContext) -> bool:
)
# 根据当前脚本位置计算资源的相对路径，便于 Notebook 中用相对路径读取文件。
@router.post("/{resource_id}/jupyter-relative-path")
async def resource_jupyter_relative_path(
resource_id: str,
@@ -163,6 +171,7 @@ async def resource_jupyter_relative_path(
}
# 上传第 1 步：创建上传会话，登记文件名、大小、类型等预期元数据。
@router.post("/uploads", status_code=status.HTTP_201_CREATED)
async def create_resource_upload(
payload: CreateResourceUploadRequest,
@@ -193,6 +202,7 @@ async def create_resource_upload(
return {"request_id": context.request_id, "data": data, "meta": {}}
# 上传第 2 步：将浏览器传来的二进制文件写入已创建的上传会话。
@router.put("/uploads/{upload_id}")
async def upload_resource_bytes(
upload_id: str,
@@ -216,6 +226,7 @@ async def upload_resource_bytes(
}
# 上传第 3 步：把已完成的上传会话绑定为工作区可见的数据资源。
@router.post("/uploads/{upload_id}/bind")
async def bind_resource(
upload_id: str,
@@ -349,6 +360,7 @@ async def bind_resource(
}
# 列出当前工作区可见的数据资源，可按可见性或关键字筛选。
@router.get("")
async def list_resources(
context: RequestContext = Depends(request_context),
@@ -428,6 +440,7 @@ async def get_visible_resource(
return row
# 查询单个数据资源的元数据与其关联文件信息。
@router.get("/{resource_id}")
async def get_resource(
resource_id: str,
@@ -446,6 +459,7 @@ async def get_resource(
}
# 为资源文件生成带时效的下载链接。
@router.post("/{resource_id}/download-url")
async def resource_download_url(
resource_id: str,
@@ -467,6 +481,7 @@ async def resource_download_url(
return {"request_id": context.request_id, "data": data["data"], "meta": {}}
# 软删除数据资源及其关联对象，遵循存储层的回收站策略。
@router.delete("/{resource_id}")
async def delete_resource(
resource_id: str,
backend/src/backend/runtime_client.py
+9
View File
@@ -1,3 +1,10 @@
"""Backend 调用 Runtime 服务的轻量 HTTP 客户端。
后端不直接管理 Jupyter 进程；涉及工作区运行时状态、编辑会话或访问票据时，
会通过本客户端请求 Docker 内网中的 ``runtime`` 服务。网络错误会统一包装成
``RuntimeClientError``，让路由返回可识别的 503 错误。
"""
from __future__ import annotations
from dataclasses import dataclass
@@ -34,6 +41,8 @@ class RuntimeClient:
path: str,
payload: dict[str, Any],
) -> dict[str, Any]:
# 将 httpx 的连接/响应异常转换为项目统一的业务异常，调用方无需关心
# 底层 HTTP 客户端的具体异常类型。
try:
response = await self.client.request(method, path, json=payload)
except httpx.RequestError as exc:
backend/src/backend/schedule_runs.py
+16
View File
@@ -1,3 +1,10 @@
"""调度运行的触发与查询 API。
手动运行接口只负责在 MySQL 中创建 ``ScheduleRuns`` 和 Outbox 事件；真正执行
任务的是 ``schedule`` 容器，它轮询 Outbox 后运行 DAG 节点。本模块还提供运行
历史、节点状态、日志和结果文件的查询入口。
"""
from __future__ import annotations
from datetime import UTC, datetime
@@ -51,6 +58,7 @@ RunStatus = Literal[
]
# 手动触发调度时允许前端附带的运行原因；严格拒绝未定义字段。
class RunScheduleRequest(StrictModel):
reason: str = Field(default="manual_run", min_length=1, max_length=255)
@@ -64,6 +72,8 @@ def _iso(value: datetime | None) -> str | None:
def _http_error_from_trigger(exc: TriggerError) -> HTTPException:
# 调度领域异常统一转换为 HTTP 状态码，前端可据此区分“找不到任务”、
# “DAG 无效”和“请求参数不合法”等情况。
if isinstance(exc, ScheduleNotFound):
return HTTPException(status.HTTP_404_NOT_FOUND, str(exc))
if isinstance(exc, InvalidDag):
@@ -226,6 +236,7 @@ async def _artifact_bytes(
)
# 立即触发一次调度：写入运行记录和 Outbox，由 schedule 容器异步接手执行。
@router.post(
"/api/v1/schedules/{schedule_id}/run",
status_code=status.HTTP_202_ACCEPTED,
@@ -276,6 +287,7 @@ async def run_schedule_now(
}
# 按调度或状态筛选运行历史，供前端运行记录列表展示。
@router.get("/api/v1/schedule-runs")
async def list_schedule_runs(
schedule_id: str | None = Query(default=None),
@@ -305,6 +317,7 @@ async def list_schedule_runs(
}
# 查询一次运行的详情，包括每个节点的执行状态。
@router.get("/api/v1/schedule-runs/{run_id}")
async def get_schedule_run(
run_id: str,
@@ -319,6 +332,7 @@ async def get_schedule_run(
}
# 返回某个节点运行关联的日志/结果产物元数据及可访问地址。
@router.get(
"/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/artifacts"
)
@@ -361,6 +375,7 @@ async def get_schedule_node_run_artifacts(
}
# 读取节点运行日志正文，通常由前端日志面板按需调用。
@router.get(
"/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/logs"
)
@@ -387,6 +402,7 @@ async def read_schedule_node_run_logs(
)
# 为节点运行结果生成下载响应或重定向地址。
@router.get(
"/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/result"
)
backend/src/backend/schedules.py
+22
View File
@@ -1,3 +1,10 @@
"""DAG 调度定义 API。
这里管理调度任务本身：名称、Cron、节点、边和 DAG 校验；实际的定时轮询与节点
执行由独立的 ``schedule`` 容器完成。``workflow_version`` 用于乐观并发控制：
前端修改画布时必须携带当前版本，避免两个人的编辑互相覆盖。
"""
from __future__ import annotations
import heapq
@@ -70,6 +77,7 @@ def cron_preview(
count: int,
base_time: datetime | None = None,
) -> dict[str, Any]:
# 该接口只计算并预览下几次触发时间，不会创建或修改任何调度任务。
if len(expression.split()) != 5:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
@@ -464,6 +472,7 @@ async def _require_valid_when_enabled(
)
# 根据 Cron 表达式预览未来触发时间，不会保存或执行任务。
@router.post("/api/v1/cron/preview")
async def preview_cron(
payload: CronPreviewRequest,
@@ -481,6 +490,7 @@ async def preview_cron(
}
# 列出调度产生的可展示版本/产物，供前端结果面板使用。
@router.get("/api/v1/schedule-artifacts")
async def list_schedule_artifacts(
limit: int = Query(default=100, ge=1, le=500),
@@ -530,6 +540,7 @@ async def list_schedule_artifacts(
}
# 列出当前工作区的调度定义及其节点、边数量等摘要信息。
@router.get("/api/v1/schedules")
async def list_schedules(
context: RequestContext = Depends(request_context),
@@ -591,6 +602,7 @@ async def list_schedules(
}
# 创建新的 DAG 调度定义；初始状态不包含节点和边。
@router.post(
"/api/v1/schedules",
status_code=status.HTTP_201_CREATED,
@@ -643,6 +655,7 @@ async def create_schedule(
}
# 获取一个调度的完整画布数据，包括节点、边和当前工作流版本。
@router.get("/api/v1/schedules/{schedule_id}")
async def get_schedule(
schedule_id: str,
@@ -657,6 +670,7 @@ async def get_schedule(
}
# 更新调度基本属性，如名称、Cron、时区、是否启用和并发策略。
@router.put("/api/v1/schedules/{schedule_id}")
@router.patch("/api/v1/schedules/{schedule_id}")
async def update_schedule(
@@ -721,6 +735,7 @@ async def update_schedule(
}
# 删除调度定义；请求携带 workflow_version 以避免误删他人刚修改的画布。
@router.delete("/api/v1/schedules/{schedule_id}")
async def delete_schedule(
schedule_id: str,
@@ -752,6 +767,7 @@ async def delete_schedule(
}
# 向调度画布新增一个执行节点，并关联已发布的脚本版本。
@router.post(
"/api/v1/schedules/{schedule_id}/nodes",
status_code=status.HTTP_201_CREATED,
@@ -807,6 +823,7 @@ async def create_schedule_node(
}
# 更新节点名称、执行参数、超时、重试和画布坐标等配置。
@router.put("/api/v1/schedules/{schedule_id}/nodes/{node_id}")
async def update_schedule_node(
schedule_id: str,
@@ -862,6 +879,7 @@ async def update_schedule_node(
}
# 从调度画布删除节点，并同步清理关联边。
@router.delete("/api/v1/schedules/{schedule_id}/nodes/{node_id}")
async def delete_schedule_node(
schedule_id: str,
@@ -915,6 +933,7 @@ async def delete_schedule_node(
}
# 在两个节点之间新增依赖边，表示目标节点必须等待源节点完成。
@router.post(
"/api/v1/schedules/{schedule_id}/edges",
status_code=status.HTTP_201_CREATED,
@@ -991,6 +1010,7 @@ async def create_schedule_edge(
}
# 修改一条依赖边的条件表达式或其他可编辑字段。
@router.put("/api/v1/schedules/{schedule_id}/edges/{edge_id}")
async def update_schedule_edge(
schedule_id: str,
@@ -1025,6 +1045,7 @@ async def update_schedule_edge(
}
# 删除节点之间的依赖关系，不会删除节点本身。
@router.delete("/api/v1/schedules/{schedule_id}/edges/{edge_id}")
async def delete_schedule_edge(
schedule_id: str,
@@ -1059,6 +1080,7 @@ async def delete_schedule_edge(
}
# 校验画布是否为可执行 DAG，例如是否存在环、孤立节点或无效版本。
@router.post("/api/v1/schedules/{schedule_id}/validate")
async def validate_schedule(
schedule_id: str,
backend/src/backend/scripts.py
+28
View File
@@ -1,3 +1,13 @@
"""工作区脚本与 Notebook 的公开 API。
脚本元数据（名称、归属、锁、版本关系）保存在 MySQL；文件正文和版本产物保存到
对象存储或本地 ``data/``。本模块负责把两者保持一致，并提供目录、上传、编辑锁、
版本发布和下载地址等接口。
前端的 ``frontend/app/services/api.ts`` 通过 ``/api/v1/scripts`` 和
``/api/v1/workspace-directories`` 调用这里的路由。
"""
import base64
import hashlib
import json
@@ -555,6 +565,7 @@ async def create_script_record(
return script, storage_object
# 新建空的 Python 脚本或 Notebook：同时创建数据库元数据和初始文件内容。
@router.post("/api/v1/scripts", status_code=status.HTTP_201_CREATED)
async def create_script(
payload: CreateScriptRequest,
@@ -587,6 +598,7 @@ async def create_script(
}
# 上传现有脚本文件：校验文件名/类型后写入存储，并建立 Scripts 记录。
@router.post(
"/api/v1/scripts/upload",
status_code=status.HTTP_201_CREATED,
@@ -647,6 +659,7 @@ async def upload_script(
}
# 返回旧版一次性完整目录树，保留给兼容旧前端；新页面通常按目录懒加载。
@router.get("/api/v1/workspace-tree")
async def get_workspace_tree(
context: RequestContext = Depends(request_context),
@@ -718,6 +731,7 @@ async def get_workspace_tree(
}
# 查询某个目录下的直接子目录，供前端按需展开工作区树。
@router.get("/api/v1/workspace-directories")
async def list_workspace_directories(
parent_path: str = Query(default=""),
@@ -790,6 +804,7 @@ async def list_workspace_directories(
}
# 在工作区内创建逻辑目录；目录信息由脚本相对路径推导，不对应容器本地文件夹。
@router.post(
"/api/v1/workspace-directories",
status_code=status.HTTP_201_CREATED,
@@ -946,6 +961,7 @@ async def create_workspace_directory(
}
# 删除逻辑目录及其下属脚本记录；实际文件按存储层的软删除规则处理。
@router.delete("/api/v1/workspace-directories")
async def delete_workspace_directory(
request: Request,
@@ -1028,6 +1044,7 @@ async def delete_workspace_directory(
}
# 列出当前工作区中用户有权查看的脚本与 Notebook 元数据。
@router.get("/api/v1/scripts")
async def list_scripts(
context: RequestContext = Depends(request_context),
@@ -1057,6 +1074,7 @@ async def list_scripts(
}
# 读取脚本正文或 Notebook JSON；编辑器打开文件时调用此接口。
@router.get("/api/v1/scripts/{script_id}/content")
async def get_script_content(
script_id: str,
@@ -1107,6 +1125,7 @@ async def get_script_content(
}
# 查询单个脚本的元数据，例如类型、路径、锁状态和拥有者。
@router.get("/api/v1/scripts/{script_id}")
async def get_script(
script_id: str,
@@ -1125,6 +1144,7 @@ async def get_script(
}
# 保存编辑器提交的新内容；会校验工作区权限和文件编辑锁。
@router.put("/api/v1/scripts/{script_id}")
async def update_script(
script_id: str,
@@ -1190,6 +1210,7 @@ async def update_script(
}
# 修改脚本锁定状态，避免其他用户同时编辑同一份文件。
@router.patch("/api/v1/scripts/{script_id}/lock")
async def set_script_lock(
script_id: str,
@@ -1232,6 +1253,7 @@ async def set_script_lock(
}
# 软删除脚本；元数据标记删除，历史版本可按规则继续保留。
@router.delete("/api/v1/scripts/{script_id}")
async def delete_script(
script_id: str,
@@ -1277,6 +1299,7 @@ async def delete_script(
}
# 将当前脚本内容发布为不可变版本，供调度节点和回溯下载使用。
@router.post(
"/api/v1/scripts/{script_id}/versions",
status_code=status.HTTP_201_CREATED,
@@ -1398,6 +1421,7 @@ async def publish_version(
}
# 列出某脚本已经发布的历史版本。
@router.get("/api/v1/scripts/{script_id}/versions")
async def list_versions(
script_id: str,
@@ -1419,6 +1443,7 @@ async def list_versions(
}
# 读取脚本最近一次发布的版本；未发布时返回空结果。
@router.get("/api/v1/scripts/{script_id}/latest-version")
async def latest_version(
script_id: str,
@@ -1470,6 +1495,7 @@ async def latest_version(
}
# 查询单个发布版本的元数据和关联脚本信息。
@router.get("/api/v1/versions/{versions_id}")
async def get_version(
versions_id: str,
@@ -1486,6 +1512,7 @@ async def get_version(
}
# 隐藏/删除一个发布版本；是否保留实际产物由存储删除策略决定。
@router.delete("/api/v1/versions/{versions_id}")
async def delete_version(
versions_id: str,
@@ -1533,6 +1560,7 @@ async def delete_version(
}
# 为某个版本产物生成带时效的下载地址，而非把大文件直接经 API 返回。
@router.post("/api/v1/versions/{versions_id}/download-url")
async def version_download_url(
versions_id: str,
backend/src/backend/storage_api.py
+14 -9
View File
@@ -1,3 +1,10 @@
"""后端内部对象存储接口与通用辅助函数。
这些路由由 ``main.py`` 额外挂载到 ``/internal``，用于上传会话、对象记录和
下载地址等内部协作。业务路由通常直接复用本模块/``services.storage`` 的函数，
而不是让浏览器直接调用这些内部接口。
"""
from __future__ import annotations
import hashlib
@@ -85,9 +92,8 @@ def safe_file_name(value: str) -> str:
return name
# Pre-resolved bucket map (for read-only callers like services/storage.py).
# Re-resolved at module load; re-resolve via resolve_bucket() if the
# workspace.artifact_bucket override matters.
# 预先解析的“用途 → 桶名”映射，供只读调用方快速使用。若工作区配置了专属
# artifact_bucket，则必须调用 resolve_bucket()，让工作区级覆盖规则生效。
BUCKET_FOR_USAGE: dict[str, str] = {
usage_type: actual_bucket_name(purpose)
for usage_type, purpose in USAGE_TYPE_TO_PURPOSE.items()
@@ -134,12 +140,9 @@ def storage_payload(item: StorageObjects) -> dict[str, Any]:
}
# Internal routes are mounted from main.py via
# ``include_router(router, prefix="/internal")``. The engine, session
# factory, and object_stores live on the main app's lifespan — this
# module only owns the route definitions and the storage-helper
# utilities (``storage_payload``, ``resolve_bucket``, ``BUCKET_FOR_USAGE``
# …) consumed by ``backend.services.storage``.
# 内部路由由 main.py 以 /internal 前缀挂载。数据库引擎、Session 工厂和对象
# 存储实例均在应用生命周期中创建；本模块只定义路由和供 services.storage 复用的
# 存储辅助函数（如 storage_payload、resolve_bucket、BUCKET_FOR_USAGE）。
router = APIRouter(tags=["internal-storage"])
@@ -470,6 +473,7 @@ async def restore_object(
}
# 管理动作：永久清理超过保留期限或指定的回收站对象。
@router.post("/v1/admin/trash/purge")
async def purge_trash_object(
payload: dict[str, Any],
@@ -520,6 +524,7 @@ async def purge_trash_object(
return {"data": {"storage_object_id": storage_object_id, "purged": True}}
# 检查内部存储后端是否可用，供健康检查和排障使用。
@router.get("/health/storage")
async def internal_health() -> dict[str, str]:
return {"status": "ready", "service": "storage-api"}
common/src/common/scheduler/trigger.py
+4 -4
View File
@@ -35,10 +35,10 @@ from common.db.models import (
from common.eventing import add_outbox_event, schedule_event_type, utcnow
from common.ids import new_ulid
# A stable user_id used for cron-triggered runs. The corresponding
# ``Users`` row is seeded by the auth-bootstrap migration so any audit
# query joining on ``ScheduleRuns.triggered_by`` still resolves.
SYSTEM_CRON_USER_ID = "01HZZZZZZZZZZZZZZZZZZZZZZCR"
# Cron 运行需要一个可审计的 ``triggered_by``。这里使用基线迁移已经创建的
# 管理员用户；该值必须是 26 个字符，才能写入 ``ScheduleRuns.triggered_by``
# 的 ``CHAR(26)`` 字段。
SYSTEM_CRON_USER_ID = "00000000000000000000000001"
TriggerType = Literal["manual", "cron", "api"]
frontend/app/features/schedules/SchedulePage.tsx
+32 -8
View File
@@ -141,19 +141,38 @@ export default function SchedulePage({
setArtifacts([]);
setSchedule(null);
setRuns([]);
setRunsLoading(true);
setRunsLoading(false);
void useSchedulesStore.getState().loadInitial();
}, [workspaceId, api, setSchedules, setArtifacts, setSchedule]);
// 进入调度页或切换调度方案后，主动加载该方案已经存在的运行记录。
// 原先只有「手动运行」和运行中的轮询会更新列表，因此从其他页面返回时会
// 一直停留在加载状态，直到产生新的运行记录。
useEffect(() => {
const scheduleId = schedule?.schedule_id;
if (!scheduleId || !api) {
setRuns([]);
setRunsLoading(false);
return;
}
void useSchedulesStore.getState().refreshRuns(scheduleId, true);
}, [schedule?.schedule_id, api, setRuns, setRunsLoading]);
useEffect(() => {
const scheduleId = schedule?.schedule_id;
if (!scheduleId || !api) return;
if (!runs.some((item) => item.run_status === "queued" || item.run_status === "running")) return;
const hasActiveRun = runs.some(
(item) => item.run_status === "queued" || item.run_status === "running",
);
const isEnabledCron = schedule.trigger_type === "cron" && schedule.enabled;
// Cron 会由后端在未来某个整分钟创建新记录。即使当前没有运行中的
// 记录，也要持续刷新，才能让新一轮运行自动出现在右侧列表中。
if (!hasActiveRun && !isEnabledCron) return;
const timer = window.setInterval(() => {
void useSchedulesStore.getState().refreshRuns(scheduleId);
}, 1500);
}, hasActiveRun ? 1500 : 3000);
return () => window.clearInterval(timer);
}, [schedule?.schedule_id, runs, api]);
}, [schedule?.schedule_id, schedule?.trigger_type, schedule?.enabled, runs, api]);
// Form submit wrapper for create-schedule modal — keeps the FormEvent flow out of JSX.
const handleCreateScheduleSubmit = (event: FormEvent<HTMLFormElement>): void => {
@@ -292,10 +311,15 @@ export default function SchedulePage({
}}
onDrop={onCanvasDrop}
onClick={(event) => {
if (event.target === event.currentTarget) {
setSelectedNodeId(null);
setSelectedEdgeId(null);
}
// 画布实际内容位于 surface/SVG 子元素中，不能只比较 currentTarget；
// 点击节点或连线时保留选择，点击其余空白位置则返回调度方案基本信息。
const target = event.target as Element;
if (
target.closest(".schedule-node") ||
target.matches(".schedule-edge-line, .schedule-edge-hit")
) return;
setSelectedNodeId(null);
setSelectedEdgeId(null);
}}
>
<div
frontend/app/features/schedules/state/schedulesStore.ts
+5 -1
View File
@@ -553,7 +553,11 @@ export const useSchedulesStore = create<State & Actions>((set, get) => {
} catch (error) {
if (showLoading) await handleError(error, "运行记录加载失败");
} finally {
if (showLoading) set({ runsLoading: false });
// 如果用户已切换到另一个调度方案，不能让旧请求结束时覆盖新方案的
// 加载状态；新方案会由自己的请求负责关闭 loading。
if (showLoading && get().schedule?.schedule_id === scheduleId) {
set({ runsLoading: false });
}
}
},
schedule/src/schedule/scheduler.py
+58 -9
View File
@@ -16,7 +16,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from datetime import UTC
from datetime import UTC, datetime, timedelta
from zoneinfo import ZoneInfo
from apscheduler.schedulers.asyncio import AsyncIOScheduler
@@ -66,6 +66,12 @@ class CronScheduler:
self._on_trigger = on_trigger
_ACTIVE_TRIGGER = on_trigger
self._sync_task: asyncio.Task[None] | None = None
# 仅在调度配置实际变化时才重置 APScheduler job。若每 5 秒都
# reschedule，一旦恰好落在整分钟之后，就可能把本分钟的触发跳过。
self._job_signatures: dict[str, tuple[str, str, int]] = {}
# APScheduler 的定时唤醒异常时，由 5 秒同步循环兜底。键保存的是
# 已由兜底路径处理的“本地整分钟”，避免同一分钟重复提交。
self._fallback_dispatched_minutes: dict[str, datetime] = {}
def start(self) -> None:
"""Start APScheduler and spawn the periodic sync loop."""
@@ -120,8 +126,9 @@ class CronScheduler:
- adds jobs for enabled cron schedules present in MySQL
- removes jobs whose schedule has been disabled / soft-deleted
- updates ``Schedules.next_run_at`` from the next APScheduler tick
- updates ``Schedules.next_run_at`` from the Cron expression itself
"""
due_schedule_ids: list[str] = []
async with session_scope(self.session_factory) as session:
schedules = list(
(
@@ -142,14 +149,16 @@ class CronScheduler:
job_id = f"schedule:{item.schedule_id}"
active_job_ids.add(job_id)
expression = (item.cron_expression or "").strip()
max_instances = max(1, item.max_concurrency)
signature = (expression, item.timezone, max_instances)
trigger = CronTrigger.from_crontab(
expression,
timezone=ZoneInfo(item.timezone),
)
if self.scheduler.get_job(job_id) is not None:
self.scheduler.reschedule_job(job_id, trigger=trigger)
updated_count += 1
else:
now = datetime.now(ZoneInfo(item.timezone))
minute = now.replace(second=0, microsecond=0)
job_changed = False
if self.scheduler.get_job(job_id) is None:
self.scheduler.add_job(
dispatch_persisted_cron,
trigger=trigger,
@@ -157,12 +166,45 @@ class CronScheduler:
id=job_id,
replace_existing=True,
coalesce=True,
max_instances=max(1, item.max_concurrency),
max_instances=max_instances,
misfire_grace_time=60,
)
added_count += 1
job = self.scheduler.get_job(job_id)
item.next_run_at = naive_utc(job.next_run_time)
job_changed = True
elif self._job_signatures.get(job_id) != signature:
# 服务重启后的首次同步也会走这里，确保持久化 job 与
# 数据库当前配置一致；之后配置不变时保留原定时点。
self.scheduler.reschedule_job(job_id, trigger=trigger)
self.scheduler.modify_job(
job_id,
max_instances=max_instances,
)
updated_count += 1
job_changed = True
self._job_signatures[job_id] = signature
# 以 CronTrigger 本身计算下次执行时间，不依赖 APScheduler 的
# 内部唤醒状态；页面展示的「下次执行」也因此保持准确。
item.next_run_at = naive_utc(
trigger.get_next_fire_time(None, now)
)
# 首次观察或刚修改表达式时，从下一个整分钟才开始兜底，符合
# Cron 的常规语义，避免用户在本分钟中途保存后立刻多跑一次。
if job_changed or job_id not in self._fallback_dispatched_minutes:
self._fallback_dispatched_minutes[job_id] = minute
# 正常情况下 APScheduler 会在整分钟回调。实测其偶发漏唤醒时，
# 这里每 5 秒检查一次当前分钟是否命中表达式，并补发一次。
due_at = trigger.get_next_fire_time(
minute - timedelta(minutes=1),
minute,
)
if (
due_at == minute
and self._fallback_dispatched_minutes.get(job_id) != minute
):
self._fallback_dispatched_minutes[job_id] = minute
due_schedule_ids.append(item.schedule_id)
removed_count = 0
for job in self.scheduler.get_jobs():
if (
@@ -170,6 +212,8 @@ class CronScheduler:
and job.id not in active_job_ids
):
self.scheduler.remove_job(job.id)
self._job_signatures.pop(job.id, None)
self._fallback_dispatched_minutes.pop(job.id, None)
removed_count += 1
if added_count or updated_count or removed_count:
logger.info(
@@ -178,6 +222,11 @@ class CronScheduler:
updated_count,
removed_count,
)
# 在数据库同步事务提交后再创建运行记录，避免两个会话同时读取调度方案
# 时发生不必要的锁等待。重复回调由运行记录的幂等键自动去重。
for schedule_id in due_schedule_ids:
logger.debug("cron fallback dispatch: schedule={}", schedule_id[-12:])
await self._on_trigger(schedule_id)
__all__ = ["CronScheduler"]