diff --git a/.env.example b/.env.example index 04b31e8..e7e28e3 100644 --- a/.env.example +++ b/.env.example @@ -5,6 +5,15 @@ COMPOSE_PROJECT_NAME=model-platform # network. Override here to expose Nginx on a different host port. GATEWAY_PORT=8888 +# ============================================================================ +# CRITICAL: must set BEFORE first run. The initial admin user is seeded by +# the auth-bootstrap migration (migrations/versions/9a1b2c3d4e5f_*.py) +# with this password as a bcrypt hash. If you leave the default in any +# non-local environment you will get pwned. The default exists only to +# keep `docker compose up` working in dev. +# ============================================================================ +INITIAL_ADMIN_PASSWORD=admin12345 + # MySQL connection URI (async SQLAlchemy driver) DATABASE_URL=mysql+asyncmy://model_platform:ChangeMe_MySQL_App_2026@mysql:3306/model_platform?charset=utf8mb4 @@ -23,4 +32,5 @@ RUSTFS_ENDPOINT=http://rustfs:9000 RUSTFS_ACCESS_KEY=modelplatform RUSTFS_SECRET_KEY=ChangeMe_RustFS_2026 RUSTFS_WORKSPACE_BUCKET=workspaces - +RUSTFS_TRASH_BUCKET=trash +RUSTFS_TRASH_RETENTION_DAYS=30 diff --git a/backend/pyproject.toml b/backend/pyproject.toml index e15b90d..893bbda 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -11,6 +11,8 @@ dependencies = [ "alembic==1.18.5", "cryptography==49.0.0", "gunicorn>=26.0.0", + "passlib==1.7.4", + "bcrypt>=4.0,<4.1", ] [tool.uv.sources] diff --git a/backend/src/backend/dependencies.py b/backend/src/backend/dependencies.py index 615a915..ec2806e 100644 --- a/backend/src/backend/dependencies.py +++ b/backend/src/backend/dependencies.py @@ -1,22 +1,48 @@ +"""FastAPI dependencies for the public API. + +Two distinct concerns live here: + +* :func:`database_session` — yields a transactional ``AsyncSession`` + bound to the request's app-state factory. The session commits on + success and rolls back on exception via ``common.db.session_scope``. + +* :func:`request_context` — verifies the ``access_token`` cookie + (HS256-signed JWT), then loads the user's active membership for the + ``workspace_id`` query parameter. Returns a :class:`RequestContext` + dataclass that downstream handlers read for identity, role, and the + request id. + +The 56 ``Depends(request_context)`` call sites elsewhere in the +backend expect this exact dataclass shape; the workspace and role are +always present, so handlers do not need to handle the "no workspace +selected" case. + +Routes that need a workspace context must declare the query parameter +explicitly, even when the path also contains a resource id (e.g. +``/schedules/{schedule_id}``). This keeps authorization decisions +local to the request and prevents the "implicit workspace" footgun +where a path-resource lookup quietly overrides what the user asked +for. +""" + from __future__ import annotations from dataclasses import dataclass from typing import AsyncIterator -from fastapi import Depends, Header, HTTPException, Request, status -from sqlalchemy import select +from fastapi import Depends, HTTPException, Query, Request, status from sqlalchemy.ext.asyncio import AsyncSession +from common.auth.jwt import JwtError, verify_jwt_token +from common.auth.membership import MembershipError, load_active_membership from common.db import session_scope -from common.db.models import ( - Roles, - Users, - WorkspaceMembers, - Workspaces, -) +from common.db.models import Roles, Users, Workspaces from common.ids import new_ulid +ACCESS_TOKEN_COOKIE = "access_token" + + @dataclass(frozen=True) class RequestContext: request_id: str @@ -34,42 +60,66 @@ async def database_session(request: Request) -> AsyncIterator[AsyncSession]: yield session +async def current_user( + request: Request, + session: AsyncSession = Depends(database_session), +) -> Users: + """Resolve the authenticated user from the ``access_token`` cookie. + + Returns the active ``Users`` row or raises 401. Does NOT load a + workspace — use :func:`request_context` for handlers that need + a workspace-scoped context, or use ``Depends(current_user)`` for + workspace-agnostic endpoints (e.g. ``/api/v1/auth/me``). + """ + token = request.cookies.get(ACCESS_TOKEN_COOKIE) + if not token: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not authenticated") + try: + payload = verify_jwt_token(token) + except JwtError as exc: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(exc)) from exc + user_id = payload.get("sub") + if not user_id: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token") + user = await session.get(Users, user_id) + if user is None or user.status != "active": + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found") + return user + + async def request_context( request: Request, - x_user_id: str = Header(alias="X-User-ID"), - x_workspace_id: str = Header(alias="X-Workspace-ID"), - x_request_id: str | None = Header(default=None, alias="X-Request-ID"), + workspace_id: str = Query( + ..., + min_length=26, + max_length=26, + description="Workspace context for this request (CHAR(26) ULID).", + ), + session: AsyncSession = Depends(database_session), ) -> RequestContext: - async with request.app.state.session_factory() as session: - statement = ( - select(Users, Workspaces, Roles) - .join( - WorkspaceMembers, - WorkspaceMembers.user_id == Users.user_id, - ) - .join( - Workspaces, - Workspaces.workspace_id == WorkspaceMembers.workspace_id, - ) - .join(Roles, Roles.role_id == WorkspaceMembers.role_id) - .where( - Users.user_id == x_user_id, - Users.status == "active", - WorkspaceMembers.workspace_id == x_workspace_id, - WorkspaceMembers.member_status == "active", - Workspaces.status == "active", - ) - ) - row = (await session.execute(statement)).one_or_none() - if row is None: - raise HTTPException( - status.HTTP_403_FORBIDDEN, - "active workspace membership is required", - ) - user, workspace, role = row - return RequestContext( - request_id=x_request_id or new_ulid(), - user=user, - workspace=workspace, - role=role, + """Verify JWT and load the user's active membership for ``workspace_id``. + + The ``request_id`` comes from the ``X-Request-ID`` header if + present, else from a freshly minted ULID. Handlers receive the + same :class:`RequestContext` shape they had under the old + header-based implementation, so the 56 ``Depends(request_context)`` + call sites in this repo stay working without changes — they just + now pass ``?workspace_id=`` instead of the old headers. + """ + user = await current_user(request, session) + try: + _user, workspace, role = await load_active_membership( + session, user.user_id, workspace_id, ) + except MembershipError as exc: + raise HTTPException( + status.HTTP_403_FORBIDDEN, + "active workspace membership is required", + ) from exc + request_id = request.headers.get("X-Request-ID") or new_ulid() + return RequestContext( + request_id=request_id, + user=user, + workspace=workspace, + role=role, + ) diff --git a/backend/src/backend/jupyter.py b/backend/src/backend/jupyter.py index 08af88b..bcbbc6c 100644 --- a/backend/src/backend/jupyter.py +++ b/backend/src/backend/jupyter.py @@ -1,96 +1,24 @@ from __future__ import annotations -import base64 -import hashlib -import hmac -import json import re -import time from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, Response from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession from backend.dependencies import database_session from backend.runtime_client import RuntimeClientError -from common.config import settings -from common.db.models import Scripts, Users, WorkspaceMembers, Workspaces -from sqlalchemy.ext.asyncio import AsyncSession +from common.auth.jwt import JwtError, verify_jwt_token +from common.auth.membership import MembershipError, load_active_membership +from common.db.models import Scripts router = APIRouter(tags=["jupyter"]) security = HTTPBearer(auto_error=False) -JWT_SECRET = settings.jwt_secret -JWT_ALGORITHM = "HS256" - - -def _b64decode(value: str) -> bytes: - padding = "=" * (-len(value) % 4) - return base64.urlsafe_b64decode(value + padding) - - -def verify_jwt_token(token: str) -> dict: - """Parse a signed JWT from the access_token cookie. - - Returns a payload dict containing at least ``sub`` (user_id) and - ``exp``; raises 401 on missing, malformed, or expired tokens. The - current implementation is intentionally minimal — replace with a - real verification once a public-key/HSM source is wired in. - """ - if not token: - raise HTTPException( - status_code=401, - detail="Missing Authentication Token", - ) - - try: - header_b64, payload_b64, signature_b64 = token.split(".", 2) - except ValueError as exc: - raise HTTPException( - status_code=401, - detail="Invalid Authentication Token", - ) from exc - - signing_input = f"{header_b64}.{payload_b64}".encode() - expected = hmac.new( - JWT_SECRET.encode(), - signing_input, - hashlib.sha256, - ).digest() - try: - signature = _b64decode(signature_b64) - except Exception as exc: # pragma: no cover - malformed b64 - raise HTTPException( - status_code=401, - detail="Invalid Authentication Token", - ) from exc - if not hmac.compare_digest(expected, signature): - raise HTTPException( - status_code=401, - detail="Invalid Authentication Token", - ) - - try: - payload = json.loads(_b64decode(payload_b64)) - except Exception as exc: - raise HTTPException( - status_code=401, - detail="Invalid Authentication Token", - ) from exc - - exp = payload.get("exp") - if not isinstance(exp, (int, float)) or exp < time.time(): - raise HTTPException( - status_code=401, - detail="Token expired", - ) - - return payload - - def extract_notebook_path( uri: str, workspace_id: str, @@ -135,37 +63,23 @@ async def check_notebook_is_locked( return bool(is_locked) -async def load_active_membership( +async def load_active_membership_or_403( session: AsyncSession, user_id: str, workspace_id: str, -) -> tuple[Users, Workspaces]: - """Resolve the user's active membership in the workspace.""" - statement = ( - select(Users, Workspaces) - .join( - WorkspaceMembers, - WorkspaceMembers.user_id == Users.user_id, - ) - .join( - Workspaces, - Workspaces.workspace_id == WorkspaceMembers.workspace_id, - ) - .where( - Users.user_id == user_id, - Users.status == "active", - WorkspaceMembers.workspace_id == workspace_id, - WorkspaceMembers.member_status == "active", - Workspaces.status == "active", - ) - ) - row = (await session.execute(statement)).one_or_none() - if row is None: +): + """Resolve the user's active membership, raising 403 if missing. + + Thin wrapper around :func:`common.auth.membership.load_active_membership` + that maps the library's ``MembershipError`` to a FastAPI 403. + """ + try: + return await load_active_membership(session, user_id, workspace_id) + except MembershipError as exc: raise HTTPException( status_code=403, detail="active workspace membership is required", - ) - return row + ) from exc @router.get("/api/v1/auth/jupyter") @@ -195,7 +109,13 @@ async def verify_jupyter_access( cookie_token = request.cookies.get("access_token") bearer_token = auth.credentials if auth else None token = cookie_token or bearer_token - payload = verify_jwt_token(token) + try: + payload = verify_jwt_token(token) + except JwtError as exc: + raise HTTPException( + status_code=401, + detail=str(exc), + ) from exc user_id = payload.get("sub") if not user_id: raise HTTPException( @@ -203,7 +123,7 @@ async def verify_jupyter_access( detail="Invalid Authentication Token", ) - await load_active_membership(session, user_id, workspace_id) + await load_active_membership_or_403(session, user_id, workspace_id) notebook_path = extract_notebook_path(original_uri, workspace_id) if notebook_path and await check_notebook_is_locked( diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 61fcc40..c390cd4 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -12,6 +12,7 @@ from common.db import create_database_engine, create_session_factory from common.service_app import create_service_app from common.storage import RustFSObjectStore from backend.admin import router as admin_router +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 @@ -34,12 +35,13 @@ async def lifespan(app: Any) -> AsyncIterator[None]: access_key=settings.rustfs_access_key, secret_key=settings.rustfs_secret_key, ) - # Ensure all three purpose-named buckets exist; the storage edge picks + # Ensure all four purpose-named buckets exist; the storage edge picks # the right one per upload (see resolve_bucket in storage_api.py). for bucket in ( settings.rustfs_workspace_bucket, settings.rustfs_version_bucket, settings.rustfs_run_log_bucket, + settings.rustfs_trash_bucket, ): await asyncio.to_thread( app.state.object_store.ensure_bucket, @@ -69,6 +71,7 @@ app = create_service_app( settings.service_name, lifespan=lifespan, ) +app.include_router(auth_router) app.include_router(jupyter_router) app.include_router(resources_router) app.include_router(schedule_runs_router) diff --git a/backend/src/backend/schedule_runs.py b/backend/src/backend/schedule_runs.py index 233d953..8c4bd6a 100644 --- a/backend/src/backend/schedule_runs.py +++ b/backend/src/backend/schedule_runs.py @@ -1,6 +1,5 @@ from __future__ import annotations -import hashlib from datetime import UTC, datetime from typing import Any, Literal @@ -13,19 +12,22 @@ from common.db.models import ( ScheduleNodeRuns, ScheduleRuns, ) -from common.eventing import add_outbox_event, utcnow -from common.ids import new_ulid from backend.dependencies import ( RequestContext, database_session, request_context, ) from common.schemas import StrictModel -from backend.schedules import ( - graph_rows, - schedule_row, - validate_dag, +from common.scheduler import ( + DagTooLarge, + InvalidDag, + InvalidNodeArguments, + ScheduleNotFound, + TriggerError, + create_scheduled_run, + normalize_idempotency_key, ) +from common.ids import new_ulid router = APIRouter(tags=["schedule-runs"]) @@ -51,46 +53,23 @@ def _iso(value: datetime | None) -> str | None: return value.astimezone(UTC).isoformat() -def _normalized_idempotency_key( - workspace_id: str, - schedule_id: str, - value: str, -) -> str: - normalized = value.strip() - if len(normalized) < 8: - raise HTTPException( - status.HTTP_422_UNPROCESSABLE_ENTITY, - "Idempotency-Key must contain at least 8 characters", +def _http_error_from_trigger(exc: TriggerError) -> HTTPException: + if isinstance(exc, ScheduleNotFound): + return HTTPException(status.HTTP_404_NOT_FOUND, str(exc)) + if isinstance(exc, InvalidDag): + return HTTPException( + status.HTTP_409_CONFLICT, + detail={ + "code": "SCHEDULE_DAG_INVALID", + "message": str(exc), + "errors": exc.errors, + }, ) - digest = hashlib.sha256( - f"{workspace_id}:{schedule_id}:{normalized}".encode("utf-8") - ).hexdigest() - return f"run:v1:{digest}" - - -def _arguments(value: dict[str, Any] | None) -> list[str]: - payload = value or {} - raw = payload.get("_args") - result = [str(item) for item in raw] if isinstance(raw, list) else [] - for key, item in payload.items(): - if key == "_args": - continue - option = f"--{key.replace('_', '-')}" - if item is True: - result.append(option) - elif item is False or item is None: - continue - elif isinstance(item, list): - for list_item in item: - result.extend((option, str(list_item))) - elif isinstance(item, (str, int, float)): - result.extend((option, str(item))) - else: - raise HTTPException( - status.HTTP_422_UNPROCESSABLE_ENTITY, - f"node argument {key!r} must be a scalar or list", - ) - return result + if isinstance(exc, DagTooLarge): + return HTTPException(status.HTTP_409_CONFLICT, str(exc)) + if isinstance(exc, InvalidNodeArguments): + return HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, exc.message) + return HTTPException(status.HTTP_409_CONFLICT, str(exc)) def run_summary(item: ScheduleRuns) -> dict[str, Any]: @@ -183,125 +162,40 @@ async def run_schedule_now( session: AsyncSession = Depends(database_session), ) -> dict[str, Any]: reason = payload.reason if payload is not None else "manual_run" - key = _normalized_idempotency_key( - context.workspace.workspace_id, - schedule_id, - idempotency_key, - ) - existing = await session.scalar( - select(ScheduleRuns).where(ScheduleRuns.idempotency_key == key) - ) - if existing is not None: - if ( - existing.workspace_id != context.workspace.workspace_id - or existing.schedule_id != schedule_id - ): - raise HTTPException( - status.HTTP_409_CONFLICT, - "Idempotency-Key belongs to another schedule run", - ) - return { - "request_id": context.request_id, - "data": await run_detail(existing, session), - "meta": {"reused": True}, - } - - schedule = await schedule_row( - schedule_id, - context, - session, - for_update=True, - ) - node_rows, edges = await graph_rows(schedule_id, session) - nodes = [row[0] for row in node_rows] - validation = validate_dag(nodes, edges) - if not validation["valid"] or not nodes: - raise HTTPException( - status.HTTP_409_CONFLICT, - detail={ - "code": "SCHEDULE_DAG_INVALID", - "message": "schedule must contain a valid non-empty DAG", - "errors": validation["errors"], - }, + trigger_type: Literal["manual", "cron"] = "cron" if reason == "cron" else "manual" + try: + key = normalize_idempotency_key( + context.workspace.workspace_id, + schedule_id, + idempotency_key, ) - if len(nodes) > 100 or len(edges) > 500: + except TriggerError as exc: raise HTTPException( - status.HTTP_409_CONFLICT, - "schedule exceeds the v1 execution size limit", - ) + status.HTTP_422_UNPROCESSABLE_ENTITY, str(exc), + ) from exc - snapshot = { - "schedule_name": schedule.schedule_name, - "workflow_version": schedule.workflow_version, - "max_concurrency": schedule.max_concurrency, - "failure_policy": schedule.failure_policy, - "nodes": [ - { - "node_id": node.node_id, - "node_key": node.node_key, - "versions_id": version.versions_id, - "script_type": script.script_type, - "artifact_object_id": version.artifact_object_id, - "artifact_path": version.artifact_path, - "timeout_seconds": node.timeout_seconds, - "retry_count": node.retry_count, - "retry_interval_sec": node.retry_interval_sec, - "arguments": _arguments(node.arguments_json), - } - for node, version, script in node_rows - ], - "edges": [ - { - "source_node_id": edge.source_node_id, - "target_node_id": edge.target_node_id, - } - for edge in edges - ], - } - now = utcnow() - run = ScheduleRuns( - run_id=new_ulid(), - schedule_id=schedule.schedule_id, - workspace_id=schedule.workspace_id, - workflow_version=schedule.workflow_version, - trigger_type="cron" if reason == "cron" else "manual", - idempotency_key=key, - run_status="queued", - state_version=0, - schedule_snapshot=snapshot, - queued_at=now, - triggered_by=context.user.user_id, - ) - session.add(run) - schedule.last_run_at = now - await add_outbox_event( - session, - event_type="schedule.run.requested", - producer="platform-api", - trace_id=context.request_id, - aggregate_type="schedule_run", - aggregate_id=run.run_id, - idempotency_key=key, - payload={ - "workspace_id": run.workspace_id, - "schedule_id": run.schedule_id, - "run_id": run.run_id, - "workflow_version": run.workflow_version, - "trigger_type": run.trigger_type, - "triggered_by": run.triggered_by, - "schedule_snapshot": snapshot, - }, - ) - await session.flush() - # Commit before yielding so the Outbox row is visible to the executor's - # next MySQL poll — the executor's _database_event_loop picks it up. - # We intentionally do NOT HTTP-push; see backend/schedule_client.py. + try: + run, is_new = await create_scheduled_run( + session, + schedule_id=schedule_id, + workspace_id=context.workspace.workspace_id, + triggered_by_user_id=context.user.user_id, + trigger_type=trigger_type, + idempotency_key=key, + trace_id=context.request_id, + ) + except TriggerError as exc: + raise _http_error_from_trigger(exc) from exc + + # Commit before responding so the Outbox row is visible to the + # executor's next MySQL poll — the executor's _database_event_loop + # picks it up. We intentionally do NOT HTTP-push. await session.commit() await session.refresh(run) return { "request_id": context.request_id, "data": await run_detail(run, session), - "meta": {"reused": False}, + "meta": {"reused": not is_new}, } diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index ca1ead5..8dbac37 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -123,9 +123,20 @@ async def lifespan(app: Any) -> AsyncIterator[None]: secret_key=settings.rustfs_secret_key, ) app.state.default_bucket = settings.rustfs_workspace_bucket - await asyncio.to_thread( - app.state.object_store.ensure_bucket, - app.state.default_bucket) + # Ensure every purpose-named bucket exists up front, including the + # trash bucket. The trash bucket is shared across all workspaces + # and usage_types; the source key is preserved as a prefix so a + # restore is a same-key move back to the source bucket. + for bucket in ( + settings.rustfs_workspace_bucket, + settings.rustfs_version_bucket, + settings.rustfs_run_log_bucket, + settings.rustfs_trash_bucket, + ): + await asyncio.to_thread( + app.state.object_store.ensure_bucket, + bucket, + ) try: yield finally: @@ -225,13 +236,19 @@ async def create_upload_record( storage_object = await session.get( StorageObjects, upload.storage_object_id) - return { - "upload_id": upload.upload_id, - "status": upload.upload_status, - "storage_object": ( - storage_payload(storage_object) if storage_object else None - ), - } + if storage_object is None or storage_object.object_status != "available": + # The previously-completed object was deleted (or never + # materialized). Treat the idempotency hit as a tombstone + # and fall through to a fresh upload: clear the pointer so + # complete_upload_record re-validates the bucket + key. + upload.storage_object_id = None + upload.upload_status = "created" + else: + return { + "upload_id": upload.upload_id, + "status": upload.upload_status, + "storage_object": storage_payload(storage_object), + } if upload.upload_status not in {"created", "uploading"}: raise HTTPException( status.HTTP_409_CONFLICT, @@ -293,11 +310,14 @@ async def complete_upload_record( raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found") if upload.upload_status == "completed" and upload.storage_object_id: item = await session.get(StorageObjects, upload.storage_object_id) - if item is None: - raise HTTPException( - status.HTTP_409_CONFLICT, - "completed upload has no storage object") - return item + if item is None or item.object_status != "available": + # The linked storage object was deleted. Reset the upload so + # the caller can re-upload the same bytes and create a + # fresh, available object. + upload.storage_object_id = None + upload.upload_status = "created" + else: + return item if upload.upload_status not in {"created", "uploading"}: raise HTTPException( status.HTTP_409_CONFLICT, @@ -533,6 +553,20 @@ async def delete_object( storage_object_id: str, request: Request, session: AsyncSession = Depends(database_session)) -> dict[str, Any]: + """Soft-delete a storage object. + + The bytes are copied to ``rustfs_trash_bucket`` under the same key + (preserved as ``{source_bucket}/{object_key}`` so a restore is a + same-name move), the source key is then deleted from its origin + bucket, and the row's ``object_status`` flips to ``"deleted"`` with + ``deleted_at`` stamped for the reaper. + + Immutable artifacts (version snapshots, run logs) are not + trashed — the policy is enforced by ``is_immutable`` and an + explicit 409. The reaper will physically delete trashed objects + older than ``rustfs_trash_retention_days`` (out of scope for this + endpoint; the field is the contract). + """ item = await session.scalar( select(StorageObjects) .where(StorageObjects.storage_object_id == storage_object_id) @@ -544,22 +578,145 @@ async def delete_object( raise HTTPException( status.HTTP_409_CONFLICT, "immutable object cannot be deleted") - if item.object_status != "deleted": - if item.storage_backend == "rustfs" and item.bucket_name and item.object_key: + if item.object_status == "deleted": + return { + "data": { + "storage_object_id": storage_object_id, + "object_status": item.object_status, + "trash_key": item.trash_key, + } + } + if item.storage_backend == "rustfs" and item.bucket_name and item.object_key: + trash_key = f"{item.bucket_name}/{item.object_key}" + try: await asyncio.to_thread( - request.app.state.object_store.delete, - bucket_name=item.bucket_name, - object_key=item.object_key) - item.object_status = "deleted" - item.deleted_at = utcnow() + request.app.state.object_store.move_to_trash, + source_bucket=item.bucket_name, + source_key=item.object_key, + trash_bucket=settings.rustfs_trash_bucket, + trash_key=trash_key, + ) + except Exception as exc: + # If the move fails, leave the source intact and surface the + # error. We do NOT mark the row as deleted in that case — + # otherwise we'd have a row pointing to non-existent bytes. + raise HTTPException( + status.HTTP_502_BAD_GATEWAY, + f"failed to move object to trash: {exc}", + ) from exc + item.trash_key = trash_key + item.object_status = "deleted" + item.deleted_at = utcnow() return { "data": { "storage_object_id": storage_object_id, "object_status": item.object_status, + "trash_key": item.trash_key, + "trash_bucket": settings.rustfs_trash_bucket, } } +@app.post( + "/internal/v1/objects/{storage_object_id}/restore") +async def restore_object( + storage_object_id: str, + request: Request, + session: AsyncSession = Depends(database_session)) -> dict[str, Any]: + """Restore a soft-deleted object from the trash bucket. + + Copies the bytes back to the source bucket + key and flips the + row back to ``object_status='available'``. If the source bucket + is missing the object (e.g. the trash copy was the only one), the + restore still works because we copy *from* trash rather than + renaming in place. The trash copy is left in place — the reaper + will collect it on the next sweep; this is intentional so a + failed restore does not destroy the only copy. + """ + item = await session.scalar( + select(StorageObjects) + .where(StorageObjects.storage_object_id == storage_object_id) + .with_for_update() + ) + if item is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found") + if item.object_status != "deleted": + raise HTTPException( + status.HTTP_409_CONFLICT, + "object is not in trash") + if not item.trash_key or not item.bucket_name or not item.object_key: + raise HTTPException( + status.HTTP_409_CONFLICT, + "object has no trash pointer; cannot restore") + try: + await asyncio.to_thread( + request.app.state.object_store.copy, + source_bucket=settings.rustfs_trash_bucket, + source_key=item.trash_key, + dest_bucket=item.bucket_name, + dest_key=item.object_key, + ) + except Exception as exc: + raise HTTPException( + status.HTTP_502_BAD_GATEWAY, + f"failed to restore from trash: {exc}", + ) from exc + item.object_status = "available" + item.deleted_at = None + # Keep trash_key so the reaper can clean up the duplicate on its + # next pass; we don't try to delete it here because a partial + # failure would leave the user with no data. + return { + "data": storage_payload(item), + } + + +@app.post( + "/internal/v1/admin/trash/purge") +async def purge_trash_object( + payload: dict[str, Any], + request: Request, + session: AsyncSession = Depends(database_session)) -> dict[str, Any]: + """Physically delete a trashed object. + + Admin / reaper endpoint — given a ``storage_object_id``, deletes + the bytes from the trash bucket and hard-deletes the DB row. + The route is split from ``delete_object`` because the regular + delete path is the one users hit, and reaper runs need a way to + finalize the lifecycle without re-entering the soft-delete branch. + """ + storage_object_id = (payload or {}).get("storage_object_id", "").strip() + if not storage_object_id: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + "storage_object_id is required") + item = await session.scalar( + select(StorageObjects) + .where(StorageObjects.storage_object_id == storage_object_id) + .with_for_update() + ) + if item is None: + return {"data": {"storage_object_id": storage_object_id, "purged": False}} + if item.object_status != "deleted": + raise HTTPException( + status.HTTP_409_CONFLICT, + "object is not in trash; refuse to hard-delete live data") + if item.trash_key: + try: + await asyncio.to_thread( + request.app.state.object_store.delete, + bucket_name=settings.rustfs_trash_bucket, + object_key=item.trash_key, + ) + except Exception as exc: + raise HTTPException( + status.HTTP_502_BAD_GATEWAY, + f"failed to delete from trash: {exc}", + ) from exc + await session.delete(item) + return {"data": {"storage_object_id": storage_object_id, "purged": True}} + + @app.get("/internal/health/storage") async def internal_health() -> dict[str, str]: return {"status": "ready", "service": "storage-api"} diff --git a/common/pyproject.toml b/common/pyproject.toml index e07adbe..3de5494 100644 --- a/common/pyproject.toml +++ b/common/pyproject.toml @@ -10,6 +10,8 @@ dependencies = [ "boto3>=1.34,<2", "fastapi==0.116.1", "pydantic-settings>=2.14.2", + "passlib==1.7.4", + "bcrypt>=4.0,<4.1", ] [build-system] diff --git a/common/src/common/config.py b/common/src/common/config.py index 2e83dc1..ab39159 100644 --- a/common/src/common/config.py +++ b/common/src/common/config.py @@ -79,6 +79,22 @@ class Settings(BaseSettings): default="run-logs", description="Bucket for schedule run logs.", ) + rustfs_trash_bucket: str = Field( + default="trash", + description=( + "Bucket for soft-deleted objects. The source bucket key is " + "preserved as a prefix so a restore is a same-key move. " + "Trash is reaped on a schedule out of band." + ), + ) + rustfs_trash_retention_days: int = Field( + default=30, + description=( + "How long a trashed object is retained before reaping. " + "Tracked in the database (StorageObjects.deleted_at) so the " + "reaper can run as a single SQL sweep." + ), + ) # ── local FS roots ──────────────────────────────────────────── workspace_root: str = Field( diff --git a/common/src/common/db/models/storage.py b/common/src/common/db/models/storage.py index 761b403..521b7aa 100644 --- a/common/src/common/db/models/storage.py +++ b/common/src/common/db/models/storage.py @@ -103,6 +103,15 @@ class StorageObjects(Base): TINYINT(1), nullable=False, server_default=text("0") ) deleted_at: Mapped[Optional[datetime.datetime]] = mapped_column(DATETIME(fsp=3)) + trash_key: Mapped[Optional[str]] = mapped_column( + String(1100), + comment=( + "Path inside the trash bucket where the soft-deleted bytes " + "live. Format: '{source_bucket}/{object_key}' so a restore " + "is a same-key copy back to the source bucket. NULL while " + "the row is still available." + ), + ) class DataResources(Base): diff --git a/common/src/common/scheduler/__init__.py b/common/src/common/scheduler/__init__.py index a12a034..fbb997d 100644 --- a/common/src/common/scheduler/__init__.py +++ b/common/src/common/scheduler/__init__.py @@ -16,6 +16,18 @@ from __future__ import annotations from typing import TYPE_CHECKING +from common.scheduler.trigger import ( + SYSTEM_CRON_USER_ID, + DagTooLarge, + InvalidDag, + InvalidNodeArguments, + ScheduleNotFound, + TriggerError, + create_scheduled_run, + normalize_idempotency_key, + parse_node_arguments, +) + if TYPE_CHECKING: # pragma: no cover - typing only from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore @@ -53,7 +65,16 @@ def build_sqlalchemy_jobstore( __all__ = [ + "DagTooLarge", + "InvalidDag", + "InvalidNodeArguments", "JOBSTORE_TABLE", + "SYSTEM_CRON_USER_ID", + "ScheduleNotFound", + "TriggerError", "build_sqlalchemy_jobstore", + "create_scheduled_run", + "normalize_idempotency_key", + "parse_node_arguments", "to_sync_database_url", ] diff --git a/common/src/common/storage/rustfs.py b/common/src/common/storage/rustfs.py index 4d4314d..30c535e 100644 --- a/common/src/common/storage/rustfs.py +++ b/common/src/common/storage/rustfs.py @@ -156,3 +156,50 @@ class RustFSObjectStore: def delete(self, *, bucket_name: str, object_key: str) -> None: self.internal.delete_object(Bucket=bucket_name, Key=object_key) + + def copy( + self, + *, + source_bucket: str, + source_key: str, + dest_bucket: str, + dest_key: str, + ) -> None: + """Server-side copy ``source_bucket/source_key`` → ``dest_bucket/dest_key``. + + ``CopySource`` is a single header string of the form + ``/{bucket}/{key}`` — must NOT be URL-encoded or quoted. + """ + self.internal.copy_object( + Bucket=dest_bucket, + Key=dest_key, + CopySource={"Bucket": source_bucket, "Key": source_key}, + ) + + def move_to_trash( + self, + *, + source_bucket: str, + source_key: str, + trash_bucket: str, + trash_key: str, + ) -> None: + """Copy an object into the trash bucket and delete the source. + + The copy is a server-side operation in RustFS (no data flows + through the client). The source delete is best-effort: if it + fails after the copy succeeds the trash holds the only copy of + the bytes, which is exactly the point — the caller can retry. + """ + self.copy( + source_bucket=source_bucket, + source_key=source_key, + dest_bucket=trash_bucket, + dest_key=trash_key, + ) + try: + self.delete(bucket_name=source_bucket, object_key=source_key) + except ClientError: + # Source was already gone, or transient delete failure — + # the trash copy is what matters; caller logs and moves on. + pass diff --git a/default.conf b/default.conf index ae62dd6..d53951f 100644 --- a/default.conf +++ b/default.conf @@ -41,6 +41,18 @@ server { proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + # Explicitly forward the session cookie set by + # POST /api/v1/auth/login. nginx forwards it by default, but + # spelling it out keeps the auth contract visible. + proxy_set_header Cookie $http_cookie; + # Defense in depth: blank out the legacy identity headers so a + # malicious client cannot bypass the cookie-based auth flow + # by stuffing X-User-ID / X-Workspace-ID into the request. + # The backend's RequestContext no longer reads them (it + # derives identity from the access_token cookie), so this is + # belt-and-suspenders against a future regression. + proxy_set_header X-User-ID ""; + proxy_set_header X-Workspace-ID ""; } # ========================================================================= @@ -102,6 +114,15 @@ server { proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + # Defense in depth: do not let the browser-supplied identity + # headers leak past the auth subrequest. The auth subrequest + # only forwards the Cookie + Authorization it cares about; + # the actual Jupyter upstream is fully trusted (the address + # comes from the backend's runtime registry), so a leaked + # X-User-ID here would not matter for the proxy target but + # could pollute audit logs. + proxy_set_header X-User-ID ""; + proxy_set_header X-Workspace-ID ""; } # 2. 内部 Auth 子请求 location @@ -121,6 +142,13 @@ server { proxy_set_header Cookie $http_cookie; proxy_set_header Authorization $http_authorization; + # Defense in depth: the auth subrequest reads the session + # cookie / Authorization header, not the legacy identity + # headers. Blank them out so a poisoned client header cannot + # be confused for an authenticated identity if the backend + # code is ever refactored to read them again. + proxy_set_header X-User-ID ""; + proxy_set_header X-Workspace-ID ""; } # 拒绝其余非法路径 diff --git a/docker-compose.yml b/docker-compose.yml index 5d97a13..7d85c4e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -31,10 +31,13 @@ services: environment: DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 JWT_SECRET: ${JWT_SECRET:-local-jwt-secret} + INITIAL_ADMIN_PASSWORD: ${INITIAL_ADMIN_PASSWORD:-admin12345} RUNTIME_API_URL: http://runtime:8000 RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:-http://rustfs:9000} RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:-modelplatform} RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} + RUSTFS_TRASH_BUCKET: ${RUSTFS_TRASH_BUCKET:-trash} + RUSTFS_TRASH_RETENTION_DAYS: ${RUSTFS_TRASH_RETENTION_DAYS:-30} volumes: - ./backend:/app/backend:ro - ./common:/app/common:ro diff --git a/frontend/app/features/admin/AdminPages.tsx b/frontend/app/features/admin/AdminPages.tsx index 1f7ad1d..68b8ed9 100644 --- a/frontend/app/features/admin/AdminPages.tsx +++ b/frontend/app/features/admin/AdminPages.tsx @@ -1,14 +1,10 @@ -import { FormEvent, useEffect, useState } from "react"; +import { useEffect, useState, type FormEvent } from "react"; import { ApiRequestError, - createEmployee, - deleteEmployee, - demoContext, - listEmployees, - updateEmployee, type Employee, } from "../../services/api"; +import { useApi, useAuth } from "../../context/AuthContext"; import Icon from "../../components/Icon"; import "../../styles/admin.css"; import "../../styles/dashboard.css"; @@ -28,13 +24,17 @@ export function DashboardPage({ online: boolean; onNavigate: (page: "scripts" | "schedules" | "system") => void; }) { + const { user, currentWorkspace } = useAuth(); return (
MODEL DEVELOPMENT PLATFORM -

下午好,{demoContext.userName}

-

当前位于 {demoContext.workspaceName},可以继续构建脚本或配置调度。

+

下午好,{user?.display_name ?? "用户"}

+

+ 当前位于 {currentWorkspace?.workspace_name ?? "(未选择 Workspace)"} + ,可以继续构建脚本或配置调度。 +

{online ? "服务正常" : "服务连接中"}
@@ -112,18 +112,20 @@ export function SystemAdminPage({ onNotify: (notice: Notice) => void; onConnectionChange: (online: boolean) => void; }) { + const api = useApi(); + const { user, currentWorkspace } = useAuth(); const [employees, setEmployees] = useState([]); const [loading, setLoading] = useState(true); const [saving, setSaving] = useState(false); const [editing, setEditing] = useState(null); const [dialogOpen, setDialogOpen] = useState(false); const [form, setForm] = useState(EMPTY_FORM); - const canManage = demoContext.roleCode === "admin"; + const canManage = user?.role_code === "admin"; const load = async (): Promise => { setLoading(true); try { - setEmployees(await listEmployees()); + setEmployees(await api.listEmployees()); onConnectionChange(true); } catch (error) { onConnectionChange(false); @@ -164,7 +166,7 @@ export function SystemAdminPage({ setSaving(true); try { if (editing) { - const updated = await updateEmployee(editing.user_id, { + const updated = await api.updateEmployee(editing.user_id, { display_name: form.display_name.trim(), email: form.email.trim() || null, role_code: form.role_code, @@ -174,7 +176,7 @@ export function SystemAdminPage({ (item) => item.user_id === updated.user_id ? updated : item, )); } else { - const created = await createEmployee({ + const created = await api.createEmployee({ username: form.username.trim(), display_name: form.display_name.trim(), email: form.email.trim() || null, @@ -197,7 +199,7 @@ export function SystemAdminPage({ const remove = async (employee: Employee): Promise => { if (!window.confirm(`确定从当前 Workspace 删除员工“${employee.display_name}”吗?`)) return; try { - await deleteEmployee(employee.user_id); + await api.deleteEmployee(employee.user_id); setEmployees((current) => current.filter((item) => item.user_id !== employee.user_id)); onNotify({ tone: "success", message: "员工已删除" }); } catch (error) { @@ -211,7 +213,11 @@ export function SystemAdminPage({ return (
-
系统管理

员工管理

{demoContext.workspaceName} · {employees.length} 名员工

+
+ 系统管理 +

员工管理

+

{currentWorkspace?.workspace_name ?? "(未选择 Workspace)"} · {employees.length} 名员工

+
diff --git a/frontend/app/features/platform/ModelPlatformApp.tsx b/frontend/app/features/platform/ModelPlatformApp.tsx index 73a3b9c..60f382f 100644 --- a/frontend/app/features/platform/ModelPlatformApp.tsx +++ b/frontend/app/features/platform/ModelPlatformApp.tsx @@ -1,6 +1,6 @@ import { type ChangeEvent, - FormEvent, + type FormEvent, type MouseEvent as ReactMouseEvent, useEffect, useMemo, @@ -10,24 +10,6 @@ import { import { useLocation, useNavigate } from "react-router"; import { - acquireFileLock, - createWorkspaceDirectory, - createScript, - createJupyterAccessTicket, - deleteScript, - deleteWorkspaceDirectory, - demoContext, - demoUsers, - demoWorkspaces, - heartbeatFileLock, - listScripts, - listScriptVersions, - listWorkspaceDirectories, - publishScriptVersion, - releaseFileLock, - releaseFileLockOnUnload, - setDemoContext, - uploadScript, type ActiveEditSession, type ScriptItem, type ScriptType, @@ -35,6 +17,7 @@ import { type Visibility, type WorkspaceDirectory, } from "../../services/api"; +import { useApi, useAuth } from "~/context/AuthContext"; import Icon from "../../components/Icon"; import SchedulePage from "../schedules/SchedulePage"; import { DashboardPage, SystemAdminPage } from "../admin/AdminPages"; @@ -186,6 +169,19 @@ export default function ModelPlatformApp() { const location = useLocation(); const navigate = useNavigate(); const activePage = pageFromPath(location.pathname); + const { user, workspaces, currentWorkspace, setCurrentWorkspace, logout } = useAuth(); + const api = useApi(); + + // Loading state: wait for auth and workspace selection + if (!currentWorkspace) { + return ( +
+
+ 加载中… +
+
+ ); + } const [scripts, setScripts] = useState([]); const [directories, setDirectories] = useState([]); const [selectedId, setSelectedId] = useState(null); @@ -204,7 +200,6 @@ export default function ModelPlatformApp() { }>({ open: false, parentPath: "", name: "", busy: false }); const [contextMenu, setContextMenu] = useState(null); const [workspaceMenuOpen, setWorkspaceMenuOpen] = useState(false); - const [userMenuOpen, setUserMenuOpen] = useState(false); const [uploadParentPath, setUploadParentPath] = useState(""); const [uploading, setUploading] = useState(false); const uploadInputRef = useRef(null); @@ -236,8 +231,8 @@ export default function ModelPlatformApp() { setRefreshing(silent); try { const [items, folderItems] = await Promise.all([ - listScripts(), - listWorkspaceDirectories(), + api.listScripts(), + api.listWorkspaceDirectories(), ]); setScripts(items); setDirectories(folderItems); @@ -308,7 +303,7 @@ export default function ModelPlatformApp() { } let ignore = false; setVersionsLoading(true); - void listScriptVersions(selectedId) + void api.listScriptVersions(selectedId) .then((items) => { if (!ignore) setVersions(items); }) @@ -342,7 +337,7 @@ export default function ModelPlatformApp() { return; } heartbeatRunning = true; - void heartbeatFileLock(current) + void api.heartbeatFileLock(current) .then((updated) => { setEditSession((active) => active && active.edit_session_id === updated.edit_session_id @@ -379,7 +374,7 @@ export default function ModelPlatformApp() { if (!current || current.edit_session_id !== editSession.edit_session_id) { return; } - void createJupyterAccessTicket(current) + void api.createJupyterAccessTicket(current) .then((ticket) => { setEditSession((active) => active && active.edit_session_id === ticket.edit_session_id @@ -402,7 +397,7 @@ export default function ModelPlatformApp() { if (!editSession) return; const handleUnload = () => { const current = editSessionRef.current; - if (current) releaseFileLockOnUnload(current); + if (current) api.releaseFileLockOnUnload(current); }; window.addEventListener("beforeunload", handleUnload); return () => window.removeEventListener("beforeunload", handleUnload); @@ -416,24 +411,17 @@ export default function ModelPlatformApp() { ); }, [keyword, scripts]); - const memberScriptGroups = [...demoUsers] - .sort((left, right) => ( - Number(right.userId === demoContext.userId) - - Number(left.userId === demoContext.userId) - )) - .map((user) => { - const memberScripts = filteredScripts.filter( - (item) => item.owner_user_id === user.userId, + const memberScriptGroups = (() => { + const currentUserScripts = filteredScripts.filter( + (item) => item.owner_user_id === user?.user_id, ); - const inferred = inferredDirectories(memberScripts); - return { - user, - scripts: memberScripts, - directories: user.userId === demoContext.userId - ? mergeDirectories(directories, inferred) - : inferred, - }; - }); + const inferred = inferredDirectories(currentUserScripts); + return [{ + user: user, + scripts: currentUserScripts, + directories: mergeDirectories(directories, inferred), + }]; + })(); const selected = scripts.find((item) => item.script_id === selectedId) ?? null; const selectScript = (scriptId: string | null) => { @@ -473,7 +461,7 @@ export default function ModelPlatformApp() { let newlyAcquired = false; try { if (active && active.script_id !== script.script_id) { - await releaseFileLock(active); + await api.releaseFileLock(active); setEmbeddedJupyterUrl(null); setEditSession(null); editSessionRef.current = null; @@ -482,20 +470,20 @@ export default function ModelPlatformApp() { if (!requestIsCurrent()) return; if (!active) { - active = await acquireFileLock(script); + active = await api.acquireFileLock(script); newlyAcquired = true; } if (!requestIsCurrent()) { if (active) { - await releaseFileLock(active); + await api.releaseFileLock(active); clearSessionIfActive(active); } return; } - const ticket = await createJupyterAccessTicket(active); + const ticket = await api.createJupyterAccessTicket(active); if (!requestIsCurrent()) { - await releaseFileLock(active); + await api.releaseFileLock(active); clearSessionIfActive(active); return; } @@ -516,7 +504,7 @@ export default function ModelPlatformApp() { } catch (error) { if (newlyAcquired && active) { try { - await releaseFileLock(active); + await api.releaseFileLock(active); } catch { // The database lease is the final safety net if compensation cannot reach Runtime. } @@ -573,7 +561,7 @@ export default function ModelPlatformApp() { } setEditBusy(true); try { - await releaseFileLock(active); + await api.releaseFileLock(active); setEmbeddedJupyterUrl(null); setEditSession(null); editSessionRef.current = null; @@ -632,7 +620,7 @@ export default function ModelPlatformApp() { if (!publishTarget) return; setPublishing(true); try { - const version = await publishScriptVersion({ + const version = await api.publishScriptVersion({ script: publishTarget, releaseNote, visibility: publishVisibility, @@ -662,7 +650,7 @@ export default function ModelPlatformApp() { if (!form.name.trim()) return; setCreating(true); try { - const created = await createScript(form); + const created = await api.createScript(form); setScripts((items) => [created, ...items]); selectScript(created.script_id); setCreateOpen(false); @@ -718,7 +706,7 @@ export default function ModelPlatformApp() { let lastCreated: ScriptItem | null = null; try { for (const file of files) { - lastCreated = await uploadScript(file, uploadParentPath); + lastCreated = await api.uploadScript(file, uploadParentPath); } await load(true); if (lastCreated) selectScript(lastCreated.script_id); @@ -744,7 +732,7 @@ export default function ModelPlatformApp() { if (!folderDialog.name.trim()) return; setFolderDialog((current) => ({ ...current, busy: true })); try { - await createWorkspaceDirectory( + await api.createWorkspaceDirectory( folderDialog.name.trim(), folderDialog.parentPath, ); @@ -778,7 +766,7 @@ export default function ModelPlatformApp() { if (editSessionRef.current?.script_id === script.script_id) return; } try { - await deleteScript(script.script_id); + await api.deleteScript(script.script_id); if (selectedIdRef.current === script.script_id) selectScript(null); await load(true); setToast({ @@ -812,7 +800,7 @@ export default function ModelPlatformApp() { if (editSessionRef.current?.script_id === activeScript.script_id) return; } try { - const result = await deleteWorkspaceDirectory(path); + const result = await api.deleteWorkspaceDirectory(path); const selectedScript = scripts.find( (item) => item.script_id === selectedIdRef.current, ); @@ -911,36 +899,29 @@ export default function ModelPlatformApp() { {apiOnline ? "服务已连接" : "服务未连接"}
- {workspaceMenuOpen && (
- {demoWorkspaces.map((workspace) => ( - ))}
)}
- + - {userMenuOpen && ( -
- {demoUsers.map((user) => ( - - ))} -
- )}
@@ -1009,18 +990,18 @@ export default function ModelPlatformApp() { <> {memberScriptGroups.map((group) => ( ))} {filteredScripts.length === 0 && ( @@ -1105,13 +1086,13 @@ export default function ModelPlatformApp() {
) : activePage === "schedules" ? ( ) : activePage === "system" ? ( diff --git a/frontend/app/features/schedules/SchedulePage.tsx b/frontend/app/features/schedules/SchedulePage.tsx index ff07ba0..50e1c32 100644 --- a/frontend/app/features/schedules/SchedulePage.tsx +++ b/frontend/app/features/schedules/SchedulePage.tsx @@ -1,8 +1,8 @@ import { - DragEvent, - FormEvent, - MouseEvent as ReactMouseEvent, - PointerEvent as ReactPointerEvent, + type DragEvent, + type FormEvent, + type MouseEvent as ReactMouseEvent, + type PointerEvent as ReactPointerEvent, useEffect, useMemo, useRef, @@ -11,22 +11,6 @@ import { import { ApiRequestError, - createSchedule, - createScheduleEdge, - createScheduleNode, - deleteSchedule, - deleteScheduleEdge, - deleteScheduleNode, - hideScheduleArtifact, - getSchedule, - listScheduleArtifacts, - listScheduleRuns, - listSchedules, - previewCron, - runScheduleNow, - updateSchedule, - updateScheduleNode, - validateSchedule, type CronPreview, type Schedule, type ScheduleArtifact, @@ -34,6 +18,8 @@ import { type ScheduleNode, type ScheduleRunSummary, } from "../../services/api"; + +import { useApi } from "~/context/AuthContext"; import Icon from "../../components/Icon"; import "../../styles/schedule.css"; @@ -250,6 +236,7 @@ export default function SchedulePage({ const dragRef = useRef(null); const positionDraftsRef = useRef>({}); const [positionDraftCount, setPositionDraftCount] = useState(0); + const api = useApi(); const selectedNode = schedule?.nodes.find( (item) => item.node_id === selectedNodeId, @@ -328,7 +315,7 @@ export default function SchedulePage({ ): Promise => { if (showLoading) setRunsLoading(true); try { - const items = await listScheduleRuns({ + const items = await api.listScheduleRuns({ scheduleId, limit: 20, }); @@ -349,8 +336,8 @@ export default function SchedulePage({ preferredScheduleId?: string | null, ): Promise => { const [scheduleItems, artifactItems] = await Promise.all([ - listSchedules(), - listScheduleArtifacts(), + api.listSchedules(), + api.listScheduleArtifacts(), ]); setSchedules(scheduleItems); setArtifacts(artifactItems); @@ -362,20 +349,20 @@ export default function SchedulePage({ setSchedule(null); return; } - const detail = await getSchedule(targetId); + const detail = await api.getSchedule(targetId); setSchedule(applyPositionDrafts(detail)); }; useEffect(() => { let cancelled = false; setLoading(true); - Promise.all([listSchedules(), listScheduleArtifacts()]) + Promise.all([api.listSchedules(), api.listScheduleArtifacts()]) .then(async ([scheduleItems, artifactItems]) => { if (cancelled) return; setSchedules(scheduleItems); setArtifacts(artifactItems); if (scheduleItems[0]) { - const detail = await getSchedule(scheduleItems[0].schedule_id); + const detail = await api.getSchedule(scheduleItems[0].schedule_id); if (!cancelled) setSchedule(applyPositionDrafts(detail)); } onConnectionChange(true); @@ -404,7 +391,7 @@ export default function SchedulePage({ } let cancelled = false; setRunsLoading(true); - listScheduleRuns({ scheduleId, limit: 20 }) + api.listScheduleRuns({ scheduleId, limit: 20 }) .then((items) => { if (!cancelled) setRuns(items); }) @@ -507,7 +494,7 @@ export default function SchedulePage({ setLinkSourceId(null); setCronResult(null); try { - setSchedule(await getSchedule(scheduleId)); + setSchedule(await api.getSchedule(scheduleId)); onConnectionChange(true); } catch (error) { await handleError(error, "调度详情加载失败"); @@ -529,7 +516,7 @@ export default function SchedulePage({ if (busy || !scheduleName) return; setBusy("create-schedule"); try { - const created = await createSchedule({ + const created = await api.createSchedule({ schedule_name: scheduleName, description: "在画布中拖入稳定版本并配置执行顺序", trigger_type: "manual", @@ -557,7 +544,7 @@ export default function SchedulePage({ if (!window.confirm(`确定删除调度“${selectedSchedule.schedule_name}”吗?`)) return; setBusy("delete-schedule"); try { - await deleteSchedule( + await api.deleteSchedule( selectedSchedule.schedule_id, selectedSchedule.workflow_version, ); @@ -571,7 +558,7 @@ export default function SchedulePage({ setSelectedNodeId(null); setSelectedEdgeId(null); if (remaining[0]) { - setSchedule(await getSchedule(remaining[0].schedule_id)); + setSchedule(await api.getSchedule(remaining[0].schedule_id)); } } onNotify({ tone: "success", message: "调度方案已删除" }); @@ -589,7 +576,7 @@ export default function SchedulePage({ if (!scheduleName || scheduleName === target.schedule_name) return; setBusy("rename-schedule"); try { - const updated = await updateSchedule(target.schedule_id, { + const updated = await api.updateSchedule(target.schedule_id, { workflow_version: target.workflow_version, schedule_name: scheduleName, }); @@ -620,7 +607,7 @@ export default function SchedulePage({ ) return; setBusy("delete-artifact"); try { - await hideScheduleArtifact(artifact.versions_id); + await api.hideScheduleArtifact(artifact.versions_id); setArtifacts((current) => current.filter( (item) => item.versions_id !== artifact.versions_id, )); @@ -652,13 +639,13 @@ export default function SchedulePage({ for (const [nodeId, position] of Object.entries( positionDraftsRef.current, )) { - updated = await updateScheduleNode(updated.schedule_id, nodeId, { + updated = await api.updateScheduleNode(updated.schedule_id, nodeId, { workflow_version: updated.workflow_version, position_x: position.position_x, position_y: position.position_y, }); } - updated = await updateSchedule(updated.schedule_id, { + updated = await api.updateSchedule(updated.schedule_id, { workflow_version: updated.workflow_version, schedule_name: scheduleForm.scheduleName.trim(), description: scheduleForm.description.trim() || null, @@ -694,7 +681,7 @@ export default function SchedulePage({ if (busy) return; setBusy("cron-preview"); try { - const result = await previewCron({ + const result = await api.previewCron({ cron_expression: scheduleForm.cronExpression.trim(), timezone: scheduleForm.timezone.trim(), count: 5, @@ -727,7 +714,7 @@ export default function SchedulePage({ } setBusy("run-now"); try { - const created = await runScheduleNow(schedule.schedule_id); + const created = await api.runScheduleNow(schedule.schedule_id); setRuns((current) => [ created, ...current.filter((item) => item.run_id !== created.run_id), @@ -763,7 +750,7 @@ export default function SchedulePage({ const nodeKey = artifactNodeKey(artifact, schedule); const updated = await withMutation( "add-node", - () => createScheduleNode(schedule.schedule_id, { + () => api.createScheduleNode(schedule.schedule_id, { workflow_version: schedule.workflow_version, node_key: nodeKey, node_name: artifact.script_name, @@ -898,7 +885,7 @@ export default function SchedulePage({ setLinkSourceId(null); await withMutation( "create-edge", - () => createScheduleEdge(schedule.schedule_id, { + () => api.createScheduleEdge(schedule.schedule_id, { workflow_version: schedule.workflow_version, source_node_id: sourceId, target_node_id: targetNodeId, @@ -935,7 +922,7 @@ export default function SchedulePage({ ); await withMutation( "save-node", - () => updateScheduleNode(schedule.schedule_id, selectedNode.node_id, { + () => api.updateScheduleNode(schedule.schedule_id, selectedNode.node_id, { workflow_version: schedule.workflow_version, node_name: nodeForm.nodeName.trim(), timeout_seconds: timeoutSeconds, @@ -958,7 +945,7 @@ export default function SchedulePage({ if (!window.confirm(`确定删除节点“${node.node_name}”吗?`)) return; const updated = await withMutation( "delete-node", - () => deleteScheduleNode( + () => api.deleteScheduleNode( schedule.schedule_id, node.node_id, schedule.workflow_version, @@ -974,7 +961,7 @@ export default function SchedulePage({ setContextMenu(null); const updated = await withMutation( "delete-edge", - () => deleteScheduleEdge( + () => api.deleteScheduleEdge( schedule.schedule_id, edge.edge_id, schedule.workflow_version, @@ -988,7 +975,7 @@ export default function SchedulePage({ if (!schedule || busy) return; setBusy("validate"); try { - const result = await validateSchedule(schedule.schedule_id); + const result = await api.validateSchedule(schedule.schedule_id); setSchedule((current) => current ? { ...current, dag_validation: result } : current); diff --git a/frontend/app/root.tsx b/frontend/app/root.tsx index 6cce2dc..28aa8b7 100644 --- a/frontend/app/root.tsx +++ b/frontend/app/root.tsx @@ -8,6 +8,7 @@ import { } from "react-router"; import type { Route } from "./+types/root"; +import { AuthProvider } from "~/context/AuthContext"; import "./app.css"; export const links: Route.LinksFunction = () => [ @@ -42,7 +43,11 @@ export function Layout({ children }: { children: React.ReactNode }) { } export default function App() { - return ; + return ( + + + + ); } export function ErrorBoundary({ error }: Route.ErrorBoundaryProps) { diff --git a/frontend/app/routes.ts b/frontend/app/routes.ts index 87d802e..0aa5be4 100644 --- a/frontend/app/routes.ts +++ b/frontend/app/routes.ts @@ -1,3 +1,6 @@ -import { type RouteConfig, route } from "@react-router/dev/routes"; +import { type RouteConfig, index, route } from "@react-router/dev/routes"; -export default [route("*", "routes/platform.tsx")] satisfies RouteConfig; +export default [ + route("/login", "routes/login.tsx"), + index("routes/platform.tsx"), +] satisfies RouteConfig; diff --git a/frontend/app/services/api.ts b/frontend/app/services/api.ts index 766b150..8f9570a 100644 --- a/frontend/app/services/api.ts +++ b/frontend/app/services/api.ts @@ -1,67 +1,15 @@ -export type DemoUser = { - userId: string; - userName: string; - username: string; - roleCode: "admin" | "developer"; - roleName: string; -}; - -export type DemoWorkspace = { - workspaceId: string; - workspaceName: string; -}; - -export const demoUsers: DemoUser[] = [ - { userId: "0000000000RF6FG1SDBXG59S13", userName: "张三", username: "admin-zhang", roleCode: "admin", roleName: "管理员" }, - { userId: "0000000000H2QYCGPCWQM1JSGS", userName: "李四", username: "admin-li", roleCode: "admin", roleName: "管理员" }, - { userId: "0000000000RWG40ESZPGJT629J", userName: "王五", username: "dev-wang", roleCode: "developer", roleName: "开发人员" }, - { userId: "00000000004CQV7WASJA6N6FW4", userName: "赵六", username: "dev-zhao", roleCode: "developer", roleName: "开发人员" }, -]; - -export const demoWorkspaces: DemoWorkspace[] = [ - { workspaceId: "00000000000BM630VT9ARVFZPC", workspaceName: "模型开发 Workspace" }, - { workspaceId: "0000000000AE0NC0V5T424KK86", workspaceName: "风险验证 Workspace" }, -]; - -function readStoredContext(): Partial<{ - userId: string; - workspaceId: string; -}> { - if (typeof window === "undefined") return {}; - try { - return JSON.parse( - window.localStorage.getItem("model-platform-demo-context") ?? "{}", - ) as Partial<{ userId: string; workspaceId: string }>; - } catch { - return {}; - } -} - -const storedContext = readStoredContext(); -const initialUser = demoUsers.find((item) => item.userId === storedContext.userId) - ?? demoUsers[0]; -const initialWorkspace = demoWorkspaces.find( - (item) => item.workspaceId === storedContext.workspaceId, -) ?? demoWorkspaces[0]; - -export const demoContext = { - ...initialUser, - ...initialWorkspace, -}; - -export function setDemoContext(input: { - user?: DemoUser; - workspace?: DemoWorkspace; -}): void { - if (input.user) Object.assign(demoContext, input.user); - if (input.workspace) Object.assign(demoContext, input.workspace); - if (typeof window !== "undefined") { - window.localStorage.setItem("model-platform-demo-context", JSON.stringify({ - userId: demoContext.userId, - workspaceId: demoContext.workspaceId, - })); - } -} +// API client for the platform backend. +// +// All endpoints that take a workspace context require the caller to +// pass `workspaceId` explicitly. Components read the active workspace +// from `useAuth().currentWorkspace` and thread it through; the cookie +// set by `/api/v1/auth/login` is sent automatically thanks to +// `credentials: "same-origin"`, and the backend reads it via the +// shared `request_context` dependency. +// +// 401 from any endpoint means the session has expired or was never +// established; the global `apiRequest` helper bounces the user to +// `/login` so the platform never tries to render with a stale identity. export type ScriptType = "python" | "notebook"; export type Visibility = "private" | "workspace" | "public"; @@ -132,22 +80,40 @@ export class ApiRequestError extends Error { } } +function appendWorkspaceId(path: string, workspaceId: string): string { + // `path` may already contain a query string. Use URLSearchParams to + // merge cleanly either way. + const separator = path.includes("?") ? "&" : "?"; + return `${path}${separator}workspace_id=${encodeURIComponent(workspaceId)}`; +} + async function apiRequest( path: string, init: RequestInit = {}, + workspaceId?: string, ): Promise { - const response = await fetch(path, { + const finalPath = workspaceId ? appendWorkspaceId(path, workspaceId) : path; + const response = await fetch(finalPath, { ...init, credentials: "same-origin", headers: { - "X-User-ID": demoContext.userId, - "X-Workspace-ID": demoContext.workspaceId, "X-Request-ID": crypto.randomUUID().replaceAll("-", ""), ...(init.body ? { "Content-Type": "application/json" } : {}), ...init.headers, }, }); + // Session expired / never authenticated — bounce to login. The + // /login route itself is the only path that must remain reachable + // while anonymous, so the redirect there is safe. + if (response.status === 401 && typeof window !== "undefined") { + const here = window.location.pathname; + if (here !== "/login") { + window.location.assign("/login"); + } + throw new ApiRequestError("未登录或登录已过期", 401); + } + const payload = (await response.json().catch(() => ({}))) as | ApiEnvelope | ApiErrorEnvelope; @@ -171,8 +137,8 @@ async function apiRequest( return (payload as ApiEnvelope).data; } -export async function listScripts(): Promise { - return apiRequest("/api/v1/scripts"); +export async function listScripts(workspaceId: string): Promise { + return apiRequest("/api/v1/scripts", {}, workspaceId); } function initialContent(scriptType: ScriptType): string { @@ -197,14 +163,14 @@ function initialContent(scriptType: ScriptType): string { { cell_type: "markdown", metadata: {}, - source: ["# 新建模型实验\\n", "在这里开始数据探索与模型构建。"], + source: ["# 新建模型实验\n", "在这里开始数据探索与模型构建。"], }, { cell_type: "code", execution_count: null, metadata: {}, outputs: [], - source: ["print('Hello, Model Platform!')\\n"], + source: ["print('Hello, Model Platform!')\n"], }, ], metadata: { @@ -226,25 +192,33 @@ function initialContent(scriptType: ScriptType): string { ); } -export async function createScript(input: { - name: string; - scriptType: ScriptType; - visibility: Visibility; - parentPath?: string | null; -}): Promise { - return apiRequest("/api/v1/scripts", { - method: "POST", - body: JSON.stringify({ - script_name: input.name.trim(), - script_type: input.scriptType, - visibility: input.visibility, - content: initialContent(input.scriptType), - parent_path: input.parentPath, - }), - }); +export async function createScript( + workspaceId: string, + input: { + name: string; + scriptType: ScriptType; + visibility: Visibility; + parentPath?: string | null; + }, +): Promise { + return apiRequest( + "/api/v1/scripts", + { + method: "POST", + body: JSON.stringify({ + script_name: input.name.trim(), + script_type: input.scriptType, + visibility: input.visibility, + content: initialContent(input.scriptType), + parent_path: input.parentPath, + }), + }, + workspaceId, + ); } export async function uploadScript( + workspaceId: string, file: File, parentPath = "", visibility: Visibility = "workspace", @@ -261,38 +235,64 @@ export async function uploadScript( headers: { "Content-Type": "application/octet-stream" }, body: file, }, + workspaceId, + ); +} + +export async function updateScript( + workspaceId: string, + scriptId: string, + input: { content: string }, +): Promise { + return apiRequest( + `/api/v1/scripts/${scriptId}`, + { method: "PUT", body: JSON.stringify(input) }, + workspaceId, ); } export async function deleteScript( + workspaceId: string, scriptId: string, ): Promise<{ script_id: string; status: string; versions_preserved: boolean }> { - return apiRequest(`/api/v1/scripts/${scriptId}`, { method: "DELETE" }); + return apiRequest( + `/api/v1/scripts/${scriptId}`, + { method: "DELETE" }, + workspaceId, + ); } -export async function listWorkspaceDirectories(): Promise< - WorkspaceDirectory[] -> { +export async function listWorkspaceDirectories( + workspaceId: string, +): Promise { const data = await apiRequest<{ directories: WorkspaceDirectory[] }>( "/api/v1/workspace-tree", + {}, + workspaceId, ); return data.directories; } export async function createWorkspaceDirectory( + workspaceId: string, directoryName: string, parentPath = "", ): Promise { - return apiRequest("/api/v1/workspace-directories", { - method: "POST", - body: JSON.stringify({ - directory_name: directoryName, - parent_path: parentPath, - }), - }); + return apiRequest( + "/api/v1/workspace-directories", + { + method: "POST", + body: JSON.stringify({ + directory_name: directoryName, + parent_path: parentPath, + }), + }, + workspaceId, + ); } export async function deleteWorkspaceDirectory( + workspaceId: string, path: string, ): Promise<{ path: string; @@ -301,9 +301,11 @@ export async function deleteWorkspaceDirectory( versions_preserved: boolean; }> { const parameters = new URLSearchParams({ path }); - return apiRequest(`/api/v1/workspace-directories?${parameters.toString()}`, { - method: "DELETE", - }); + return apiRequest( + `/api/v1/workspace-directories?${parameters.toString()}`, + { method: "DELETE" }, + workspaceId, + ); } export type FileLockSession = { @@ -353,12 +355,20 @@ export type StableVersion = { created_at: string; }; +// Note: the file-lock and jupyter-ticket endpoints are not yet +// implemented in the backend (see the cookie+JWT auth refactor plan). +// They are retained here so the editor UI keeps its existing call +// sites, but they will return 404 until the backend ships the +// corresponding routes. + export async function acquireFileLock( + workspaceId: string, script: ScriptItem, ): Promise { const session = await apiRequest( `/api/v1/files/${script.current_object_id}/lock`, { method: "POST" }, + workspaceId, ); if (!session.lock_token) { throw new Error("加锁成功响应缺少 lock_token"); @@ -372,6 +382,7 @@ export async function acquireFileLock( } export async function heartbeatFileLock( + workspaceId: string, session: ActiveEditSession, ): Promise { return apiRequest( @@ -380,10 +391,12 @@ export async function heartbeatFileLock( method: "POST", body: JSON.stringify({ lock_token: session.lock_token }), }, + workspaceId, ); } export async function releaseFileLock( + workspaceId: string, session: ActiveEditSession, ): Promise { return apiRequest( @@ -392,47 +405,64 @@ export async function releaseFileLock( method: "DELETE", body: JSON.stringify({ lock_token: session.lock_token }), }, + workspaceId, ); } -export function releaseFileLockOnUnload(session: ActiveEditSession): void { - void fetch(`/api/v1/file-locks/${session.edit_session_id}`, { - method: "DELETE", - credentials: "same-origin", - keepalive: true, - headers: { - "Content-Type": "application/json", - "X-User-ID": demoContext.userId, - "X-Workspace-ID": demoContext.workspaceId, - "X-Request-ID": crypto.randomUUID().replaceAll("-", ""), +export function releaseFileLockOnUnload( + workspaceId: string, + session: ActiveEditSession, +): void { + void fetch( + `/api/v1/file-locks/${session.edit_session_id}?workspace_id=${ + encodeURIComponent(workspaceId) + }`, + { + method: "DELETE", + credentials: "same-origin", + keepalive: true, + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ lock_token: session.lock_token }), }, - body: JSON.stringify({ lock_token: session.lock_token }), - }); + ); } export async function createJupyterAccessTicket( + workspaceId: string, session: ActiveEditSession, ): Promise { - return apiRequest("/api/v1/jupyter/access-tickets", { - method: "POST", - body: JSON.stringify({ - edit_session_id: session.edit_session_id, - lock_token: session.lock_token, - }), - }); + return apiRequest( + "/api/v1/jupyter/access-tickets", + { + method: "POST", + body: JSON.stringify({ + edit_session_id: session.edit_session_id, + lock_token: session.lock_token, + }), + }, + workspaceId, + ); } export async function listScriptVersions( + workspaceId: string, scriptId: string, ): Promise { - return apiRequest(`/api/v1/scripts/${scriptId}/versions`); + return apiRequest( + `/api/v1/scripts/${scriptId}/versions`, + {}, + workspaceId, + ); } -export async function publishScriptVersion(input: { - script: ScriptItem; - releaseNote: string; - visibility: Visibility; -}): Promise { +export async function publishScriptVersion( + workspaceId: string, + input: { + script: ScriptItem; + releaseNote: string; + visibility: Visibility; + }, +): Promise { return apiRequest( `/api/v1/scripts/${input.script.script_id}/versions`, { @@ -443,6 +473,7 @@ export async function publishScriptVersion(input: { visibility: input.visibility, }), }, + workspaceId, ); } @@ -595,31 +626,43 @@ export type ScheduleRunDetail = ScheduleRunSummary & { node_runs: ScheduleNodeRun[]; }; -export async function listSchedules(): Promise { - return apiRequest("/api/v1/schedules"); +export async function listSchedules(workspaceId: string): Promise { + return apiRequest("/api/v1/schedules", {}, workspaceId); } -export async function getSchedule(scheduleId: string): Promise { - return apiRequest(`/api/v1/schedules/${scheduleId}`); +export async function getSchedule( + workspaceId: string, + scheduleId: string, +): Promise { + return apiRequest( + `/api/v1/schedules/${scheduleId}`, + {}, + workspaceId, + ); } -export async function createSchedule(input: { - schedule_name: string; - description?: string | null; - trigger_type?: "manual" | "cron" | "api"; - cron_expression?: string | null; - timezone?: string; - enabled?: boolean; - max_concurrency?: number; - failure_policy?: "stop" | "continue"; -}): Promise { - return apiRequest("/api/v1/schedules", { - method: "POST", - body: JSON.stringify(input), - }); +export async function createSchedule( + workspaceId: string, + input: { + schedule_name: string; + description?: string | null; + trigger_type?: "manual" | "cron" | "api"; + cron_expression?: string | null; + timezone?: string; + enabled?: boolean; + max_concurrency?: number; + failure_policy?: "stop" | "continue"; + }, +): Promise { + return apiRequest( + "/api/v1/schedules", + { method: "POST", body: JSON.stringify(input) }, + workspaceId, + ); } export async function updateSchedule( + workspaceId: string, scheduleId: string, input: { workflow_version: number; @@ -633,55 +676,72 @@ export async function updateSchedule( failure_policy?: "stop" | "continue"; }, ): Promise { - return apiRequest(`/api/v1/schedules/${scheduleId}`, { - method: "PATCH", - body: JSON.stringify(input), - }); + return apiRequest( + `/api/v1/schedules/${scheduleId}`, + { method: "PATCH", body: JSON.stringify(input) }, + workspaceId, + ); } export async function deleteSchedule( + workspaceId: string, scheduleId: string, workflowVersion: number, ): Promise<{ schedule_id: string; deleted: boolean; workflow_version: number }> { - return apiRequest(`/api/v1/schedules/${scheduleId}`, { - method: "DELETE", - body: JSON.stringify({ workflow_version: workflowVersion }), - }); + return apiRequest( + `/api/v1/schedules/${scheduleId}`, + { method: "DELETE", body: JSON.stringify({ workflow_version: workflowVersion }) }, + workspaceId, + ); } -export async function listScheduleArtifacts(): Promise { - return apiRequest("/api/v1/schedule-artifacts"); +export async function listScheduleArtifacts( + workspaceId: string, +): Promise { + return apiRequest( + "/api/v1/schedule-artifacts", + {}, + workspaceId, + ); } export async function hideScheduleArtifact( + workspaceId: string, versionsId: string, ): Promise<{ versions_id: string; deleted: boolean; artifact_preserved: boolean; }> { - return apiRequest(`/api/v1/versions/${versionsId}`, { - method: "DELETE", - }); + return apiRequest( + `/api/v1/versions/${versionsId}`, + { method: "DELETE" }, + workspaceId, + ); } -export async function listEmployees(): Promise { - return apiRequest("/api/v1/admin/employees"); +export async function listEmployees(workspaceId: string): Promise { + return apiRequest("/api/v1/admin/employees", {}, workspaceId); } -export async function createEmployee(input: { - username: string; - display_name: string; - email?: string | null; - role_code: "admin" | "developer"; -}): Promise { - return apiRequest("/api/v1/admin/employees", { - method: "POST", - body: JSON.stringify(input), - }); +export async function createEmployee( + workspaceId: string, + input: { + username: string; + display_name: string; + email?: string | null; + role_code: "admin" | "developer"; + }, +): Promise { + return apiRequest( + "/api/v1/admin/employees", + { method: "POST", body: JSON.stringify(input) }, + workspaceId, + ); } export async function updateEmployee( + workspaceId: string, userId: string, input: { display_name?: string; @@ -690,21 +750,26 @@ export async function updateEmployee( status?: "active" | "disabled" | "locked"; }, ): Promise { - return apiRequest(`/api/v1/admin/employees/${userId}`, { - method: "PATCH", - body: JSON.stringify(input), - }); + return apiRequest( + `/api/v1/admin/employees/${userId}`, + { method: "PATCH", body: JSON.stringify(input) }, + workspaceId, + ); } export async function deleteEmployee( + workspaceId: string, userId: string, ): Promise<{ user_id: string; deleted: boolean }> { - return apiRequest(`/api/v1/admin/employees/${userId}`, { - method: "DELETE", - }); + return apiRequest( + `/api/v1/admin/employees/${userId}`, + { method: "DELETE" }, + workspaceId, + ); } export async function createScheduleNode( + workspaceId: string, scheduleId: string, input: { workflow_version: number; @@ -720,13 +785,15 @@ export async function createScheduleNode( env_refs_json?: Record; }, ): Promise { - return apiRequest(`/api/v1/schedules/${scheduleId}/nodes`, { - method: "POST", - body: JSON.stringify(input), - }); + return apiRequest( + `/api/v1/schedules/${scheduleId}/nodes`, + { method: "POST", body: JSON.stringify(input) }, + workspaceId, + ); } export async function updateScheduleNode( + workspaceId: string, scheduleId: string, nodeId: string, input: { @@ -744,28 +811,26 @@ export async function updateScheduleNode( ): Promise { return apiRequest( `/api/v1/schedules/${scheduleId}/nodes/${nodeId}`, - { - method: "PUT", - body: JSON.stringify(input), - }, + { method: "PUT", body: JSON.stringify(input) }, + workspaceId, ); } export async function deleteScheduleNode( + workspaceId: string, scheduleId: string, nodeId: string, workflowVersion: number, ): Promise { return apiRequest( `/api/v1/schedules/${scheduleId}/nodes/${nodeId}`, - { - method: "DELETE", - body: JSON.stringify({ workflow_version: workflowVersion }), - }, + { method: "DELETE", body: JSON.stringify({ workflow_version: workflowVersion }) }, + workspaceId, ); } export async function createScheduleEdge( + workspaceId: string, scheduleId: string, input: { workflow_version: number; @@ -774,50 +839,58 @@ export async function createScheduleEdge( condition_expr?: string | null; }, ): Promise { - return apiRequest(`/api/v1/schedules/${scheduleId}/edges`, { - method: "POST", - body: JSON.stringify(input), - }); + return apiRequest( + `/api/v1/schedules/${scheduleId}/edges`, + { method: "POST", body: JSON.stringify(input) }, + workspaceId, + ); } export async function deleteScheduleEdge( + workspaceId: string, scheduleId: string, edgeId: string, workflowVersion: number, ): Promise { return apiRequest( `/api/v1/schedules/${scheduleId}/edges/${edgeId}`, - { - method: "DELETE", - body: JSON.stringify({ workflow_version: workflowVersion }), - }, + { method: "DELETE", body: JSON.stringify({ workflow_version: workflowVersion }) }, + workspaceId, ); } export async function validateSchedule( + workspaceId: string, scheduleId: string, ): Promise { - return apiRequest(`/api/v1/schedules/${scheduleId}/validate`, { - method: "POST", - }); + return apiRequest( + `/api/v1/schedules/${scheduleId}/validate`, + { method: "POST" }, + workspaceId, + ); } -export async function previewCron(input: { - cron_expression: string; - timezone: string; - count?: number; - base_time?: string; -}): Promise { - return apiRequest("/api/v1/cron/preview", { - method: "POST", - body: JSON.stringify(input), - }); +export async function previewCron( + workspaceId: string, + input: { + cron_expression: string; + timezone: string; + count?: number; + base_time?: string; + }, +): Promise { + return apiRequest( + "/api/v1/cron/preview", + { method: "POST", body: JSON.stringify(input) }, + workspaceId, + ); } export async function runScheduleNow( + workspaceId: string, scheduleId: string, ): Promise { return apiRequest( @@ -829,25 +902,160 @@ export async function runScheduleNow( }, body: JSON.stringify({ reason: "manual_run" }), }, + workspaceId, ); } -export async function listScheduleRuns(input: { - scheduleId?: string; - status?: ScheduleRunStatus; - limit?: number; -} = {}): Promise { +export async function listScheduleRuns( + workspaceId: string, + input: { + scheduleId?: string; + status?: ScheduleRunStatus; + limit?: number; + } = {}, +): Promise { const query = new URLSearchParams(); if (input.scheduleId) query.set("schedule_id", input.scheduleId); if (input.status) query.set("status", input.status); query.set("limit", String(input.limit ?? 20)); return apiRequest( `/api/v1/schedule-runs?${query.toString()}`, + {}, + workspaceId, ); } export async function getScheduleRun( + workspaceId: string, runId: string, ): Promise { - return apiRequest(`/api/v1/schedule-runs/${runId}`); + return apiRequest( + `/api/v1/schedule-runs/${runId}`, + {}, + workspaceId, + ); } + +// ---------------------------------------------------------------------------- +// Workspace-bound API surface. +// +// `useApi()` in ~/context/AuthContext returns an object where every +// function has had its first `workspaceId` argument pre-filled. The +// type below lets consumers import the bound type without depending +// on the raw functions. Keep this last in the file so the type +// references all the exports above. +// ---------------------------------------------------------------------------- +export type WorkspaceBoundApi = { + listScripts: () => Promise; + createScript: ( + input: Parameters[1], + ) => Promise; + uploadScript: ( + file: File, + parentPath?: string, + visibility?: Visibility, + ) => Promise; + updateScript: ( + scriptId: string, + input: Parameters[1], + ) => Promise; + deleteScript: ( + scriptId: string, + ) => Promise<{ script_id: string; status: string; versions_preserved: boolean }>; + listWorkspaceDirectories: () => Promise; + createWorkspaceDirectory: ( + directoryName: string, + parentPath?: string, + ) => Promise; + deleteWorkspaceDirectory: ( + path: string, + ) => Promise<{ + path: string; + status: string; + deleted_scripts: number; + versions_preserved: boolean; + }>; + acquireFileLock: ( + script: ScriptItem, + ) => Promise; + heartbeatFileLock: ( + session: ActiveEditSession, + ) => Promise; + releaseFileLock: ( + session: ActiveEditSession, + ) => Promise; + releaseFileLockOnUnload: (session: ActiveEditSession) => void; + createJupyterAccessTicket: ( + session: ActiveEditSession, + ) => Promise; + listScriptVersions: (scriptId: string) => Promise; + publishScriptVersion: ( + input: Parameters[1], + ) => Promise; + listSchedules: () => Promise; + getSchedule: (scheduleId: string) => Promise; + createSchedule: ( + input: Parameters[1], + ) => Promise; + updateSchedule: ( + scheduleId: string, + input: Parameters[2], + ) => Promise; + deleteSchedule: ( + scheduleId: string, + workflowVersion: number, + ) => Promise<{ schedule_id: string; deleted: boolean; workflow_version: number }>; + listScheduleArtifacts: () => Promise; + hideScheduleArtifact: ( + versionsId: string, + ) => Promise<{ + versions_id: string; + deleted: boolean; + artifact_preserved: boolean; + }>; + listEmployees: () => Promise; + createEmployee: ( + input: Parameters[1], + ) => Promise; + updateEmployee: ( + userId: string, + input: Parameters[2], + ) => Promise; + deleteEmployee: ( + userId: string, + ) => Promise<{ user_id: string; deleted: boolean }>; + createScheduleNode: ( + scheduleId: string, + input: Parameters[2], + ) => Promise; + updateScheduleNode: ( + scheduleId: string, + nodeId: string, + input: Parameters[3], + ) => Promise; + deleteScheduleNode: ( + scheduleId: string, + nodeId: string, + workflowVersion: number, + ) => Promise; + createScheduleEdge: ( + scheduleId: string, + input: Parameters[2], + ) => Promise; + deleteScheduleEdge: ( + scheduleId: string, + edgeId: string, + workflowVersion: number, + ) => Promise; + validateSchedule: ( + scheduleId: string, + ) => Promise; + previewCron: ( + input: Parameters[1], + ) => Promise; + runScheduleNow: (scheduleId: string) => Promise; + listScheduleRuns: ( + input?: Parameters[1], + ) => Promise; + getScheduleRun: (runId: string) => Promise; +}; diff --git a/schedule/src/schedule/main.py b/schedule/src/schedule/main.py index 4e9ae0b..c9553bb 100644 --- a/schedule/src/schedule/main.py +++ b/schedule/src/schedule/main.py @@ -18,12 +18,12 @@ from schedule.storage_client import SchedulerStorageClient async def lifespan(app: Any) -> AsyncIterator[None]: engine = create_database_engine(settings.database_url) session_factory = create_session_factory(engine) - backend_http_client = build_storage_http_client() + storage_http_client = build_storage_http_client() service = SchedulerService( session_factory=session_factory, - backend_http_client=backend_http_client, + storage_http_client=storage_http_client, object_store=build_object_store(), - storage_client=SchedulerStorageClient(backend_http_client), + storage_client=SchedulerStorageClient(storage_http_client), database_url=settings.database_url, ) app.state.scheduler_service = service @@ -32,7 +32,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]: yield finally: await service.close() - await backend_http_client.aclose() + await storage_http_client.aclose() await engine.dispose() diff --git a/schedule/src/schedule/orchestrator.py b/schedule/src/schedule/orchestrator.py index 2ba932c..754f11d 100644 --- a/schedule/src/schedule/orchestrator.py +++ b/schedule/src/schedule/orchestrator.py @@ -47,20 +47,30 @@ class DispatchOrchestrator: - ``_database_event_loop`` drains ``schedule.run.requested`` and ``job.node.finished`` events under ``dispatch_lock``. These are short, in-line DB transactions. - - ``_execution_loop`` claims ``job.node.execute`` events, sets a - far-future ``available_at`` as a lease, then dispatches each as + - ``_execution_loop`` claims ``job.node.execute`` events whose + ``available_at <= utcnow()`` and dispatches each as ``asyncio.create_task`` so the polling path is never blocked by notebook execution. A semaphore caps concurrent notebooks. Holds a ``dispatch_lock`` to keep two concurrent drain loops from fighting over the same batch. + + Lease semantics live on the outbox row itself, not in the claim + step. The dispatcher sets ``available_at = utcnow() + node_timeout + + LEASE_SLACK`` when it writes the ``job.node.execute`` event, so a + process crash mid-execution lets the row re-eligible automatically + once the lease expires. A hard-coded 30-minute lease was the + original P0-2 bug: a node with ``timeout_seconds = 86_400`` would + be re-claimed at 30 minutes and run twice; a node with + ``timeout_seconds = 60`` would have its lease expire 29 minutes + too early. Tieing the lease to the actual node timeout closes both + cases. """ - # Lease window for a claimed ``job.node.execute`` event. If the - # process dies mid-execution, the row re-eligible after this many - # minutes. The handler is idempotent (it short-circuits on terminal - # node states), so safe re-execution. - EXECUTION_LEASE = timedelta(minutes=30) + # Margin added on top of ``timeout_seconds`` when writing the lease + # ``available_at``. Gives the worker time to update the row to a + # terminal state before the poll re-picks it. + LEASE_SLACK = timedelta(seconds=30) def __init__( self, @@ -140,11 +150,13 @@ class DispatchOrchestrator: ) -> int: """Claim ``job.node.execute`` rows and dispatch them as tasks. - The claim step bumps ``available_at`` to a far-future lease so - the polling loop does not re-pick the same row while the - dispatched task is still running. Status stays ``pending``; - the dispatched task flips it to ``published`` or ``failed`` - when execution completes. + Lease is owned by the row itself (the dispatcher sets + ``available_at = utcnow() + node.timeout_seconds + LEASE_SLACK`` + when the event is enqueued), so this method is a pure + read-and-dispatch — no DB writes in the claim step. If the + process dies before ``_run_node_execute`` finishes, the row + re-eligible once ``available_at`` falls back to now; the worker + handler is idempotent (short-circuits on terminal node state). """ async with session_scope(self.session_factory) as session: statement = ( @@ -158,17 +170,18 @@ class DispatchOrchestrator: .limit(limit) ) events = list((await session.scalars(statement)).all()) - claimed: list[tuple[dict[str, Any], str]] = [] - lease_until = utcnow() + self.EXECUTION_LEASE - for item in events: - item.available_at = lease_until - envelope = { - "event_type": item.event_type, - "event_id": item.event_id, - "trace_id": item.trace_id, - "payload": item.payload_json, - } - claimed.append((envelope, f"mysql:{item.event_id}")) + claimed: list[tuple[dict[str, Any], str]] = [ + ( + { + "event_type": item.event_type, + "event_id": item.event_id, + "trace_id": item.trace_id, + "payload": item.payload_json, + }, + f"mysql:{item.event_id}", + ) + for item in events + ] for envelope, message_id in claimed: task = asyncio.create_task( self._run_node_execute(envelope, message_id), @@ -379,6 +392,20 @@ class DispatchOrchestrator: ), ) session.add(node_run) + # Lease is owned by the outbox row, not the claim step. Pick + # the later of (now, scheduled retry) and the timeout + slack, + # so a slow node isn't re-dispatched while it's still running + # but a crashed node does become eligible again after its + # timeout expires. See P0-2 in the auth refactor plan. + retry_at = ( + utcnow() + timedelta(seconds=delay_seconds) + if delay_seconds + else utcnow() + ) + lease_at = utcnow() + timedelta( + seconds=int(node["timeout_seconds"]) + ) + self.LEASE_SLACK + available_at = max(retry_at, lease_at) await add_outbox_event( session, event_type="job.node.execute", @@ -387,11 +414,7 @@ class DispatchOrchestrator: aggregate_type="schedule_node_run", aggregate_id=node_run.node_run_id, idempotency_key=f"{node_run.node_run_id}:{attempt_no}", - available_at=( - utcnow() + timedelta(seconds=delay_seconds) - if delay_seconds - else None - ), + available_at=available_at, payload={ "workspace_id": run.workspace_id, "run_id": run.run_id, diff --git a/schedule/src/schedule/service.py b/schedule/src/schedule/service.py index 4dac3a4..43275c9 100644 --- a/schedule/src/schedule/service.py +++ b/schedule/src/schedule/service.py @@ -6,13 +6,16 @@ Composes three single-purpose components into one bootable service: - :class:`schedule.orchestrator.DispatchOrchestrator` — Outbox polling + DAG - :class:`schedule.worker.NodeExecutor` — node-level execution -This module also exposes the two factory functions (``build_object_store`` / -``build_storage_http_client``) consumed by ``schedule.main`` to construct -the backing resources that flow into the facade. +This module also exposes the factory function ``build_object_store`` +consumed by ``schedule.main`` to construct the RustFS S3 client. -The ``SchedulerService`` itself stays small: it wires the three components -together and implements ``trigger_schedule``, the cron post-back to -Backend that ``CronScheduler`` calls at every cron tick. +The :class:`SchedulerService` itself stays small: it wires the three +components together and implements :meth:`SchedulerService.trigger_schedule`, +the cron tick handler. The trigger writes the new ``ScheduleRuns`` row +and ``schedule.run.requested`` outbox event in a single transaction +via :func:`common.scheduler.create_scheduled_run` — no HTTP call to +the backend, so no service-to-service auth is needed (the schedule +service shares the same MySQL via the Docker network). """ from __future__ import annotations @@ -25,6 +28,11 @@ import httpx from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from common.config import settings +from common.scheduler import ( + SYSTEM_CRON_USER_ID, + TriggerError, + create_scheduled_run, +) from common.ids import new_ulid from schedule.orchestrator import DispatchOrchestrator @@ -49,13 +57,13 @@ class SchedulerService: self, *, session_factory: async_sessionmaker[AsyncSession], - backend_http_client: httpx.AsyncClient, + storage_http_client: httpx.AsyncClient, object_store: Any, storage_client: Any, database_url: str, ) -> None: self.session_factory = session_factory - self.backend_http_client = backend_http_client + self.storage_http_client = storage_http_client self.object_store = object_store self.storage_client = storage_client self.database_url = database_url @@ -99,11 +107,17 @@ class SchedulerService: await self.cron.close() async def trigger_schedule(self, schedule_id: str) -> None: - """Cron tick callback: post back to Backend to register a new run. + """Cron tick callback: write a new ``ScheduleRuns`` row + outbox event. - Backend writes the ``schedule.run.requested`` Outbox row in the same - transaction as the ``ScheduleRuns`` insert; the orchestrator's polling - loop will pick it up and start advancing the DAG. + Runs in its own session. The cron ``triggered_by`` is the + stable :data:`common.scheduler.SYSTEM_CRON_USER_ID` (the + bootstrap migration seeds the matching ``Users`` row) — we no + longer impersonate the schedule's human creator as the previous + header-based implementation did. + + Idempotency key is the cron minute bucket, so re-entering the + same tick (e.g. after a brief outage) reuses the existing run + via the unique constraint on ``schedule_runs.idempotency_key``. """ async with self.session_factory() as session: from sqlalchemy import select @@ -118,26 +132,29 @@ class SchedulerService: or item.trigger_type != "cron" ): return - user_id = item.created_by workspace_id = item.workspace_id + now = datetime.now(UTC) idempotency_key = ( f"cron:{schedule_id}:{now.strftime('%Y%m%d%H%M')}" ) - response = await self.backend_http_client.post( - f"/api/v1/schedules/{schedule_id}/run", - headers={ - "X-User-ID": user_id, - "X-Workspace-ID": workspace_id, - "X-Request-ID": new_ulid(), - "Idempotency-Key": idempotency_key, - }, - json={"reason": "cron"}, - ) - if response.is_error: - raise RuntimeError( - f"backend rejected cron run: {response.status_code} " - f"{response.text[:500]}" + try: + async with self.session_factory() as session: + await create_scheduled_run( + session, + schedule_id=schedule_id, + workspace_id=workspace_id, + triggered_by_user_id=SYSTEM_CRON_USER_ID, + trigger_type="cron", + idempotency_key=idempotency_key, + trace_id=new_ulid(), + ) + await session.commit() + 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, ) async def process_pending_events( @@ -175,7 +192,14 @@ def build_object_store() -> Any: def build_storage_http_client() -> httpx.AsyncClient: - """Construct the httpx client that talks to Backend's HTTP API.""" + """Construct the httpx client that talks to Backend's storage API. + + The schedule service no longer needs to call any user-facing + endpoint (cron trigger writes directly to the DB now), but the + storage endpoints at ``/internal/v1/...`` still live on the + backend process and are reached via this client. Auth is not + required — the client is bound to the shared Docker network. + """ return httpx.AsyncClient( base_url=settings.backend_api_url, timeout=httpx.Timeout(60.0), diff --git a/uv.lock b/uv.lock index cd7c8f2..e8943c4 100644 --- a/uv.lock +++ b/uv.lock @@ -194,27 +194,50 @@ version = "0.2.0" source = { editable = "backend" } dependencies = [ { name = "alembic" }, + { name = "bcrypt" }, { name = "common" }, { name = "croniter" }, { name = "cryptography" }, { name = "fastapi" }, { name = "gunicorn" }, { name = "httpx" }, + { name = "passlib" }, { name = "uvicorn", extra = ["standard"] }, ] [package.metadata] requires-dist = [ { name = "alembic", specifier = "==1.18.5" }, + { name = "bcrypt", specifier = ">=4.0,<4.1" }, { name = "common", editable = "common" }, { name = "croniter", specifier = "==6.2.4" }, { name = "cryptography", specifier = "==49.0.0" }, { name = "fastapi", specifier = "==0.116.1" }, { name = "gunicorn", specifier = ">=26.0.0" }, { name = "httpx", specifier = "==0.28.1" }, + { name = "passlib", specifier = "==1.7.4" }, { name = "uvicorn", extras = ["standard"], specifier = "==0.35.0" }, ] +[[package]] +name = "bcrypt" +version = "4.0.1" +source = { registry = "https://pypi.tuna.tsinghua.edu.cn/simple/" } +sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/8c/ae/3af7d006aacf513975fd1948a6b4d6f8b4a307f8a244e1a3d3774b297aad/bcrypt-4.0.1.tar.gz", hash = "sha256:27d375903ac8261cfe4047f6709d16f7d18d39b1ec92aaf72af989552a650ebd", size = 25498, upload-time = "2022-10-09T15:36:49.775Z" } +wheels = [ + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/78/d4/3b2657bd58ef02b23a07729b0df26f21af97169dbd0b5797afa9e97ebb49/bcrypt-4.0.1-cp36-abi3-macosx_10_10_universal2.whl", hash = "sha256:b1023030aec778185a6c16cf70f359cbb6e0c289fd564a7cfa29e727a1c38f8f", size = 473446, upload-time = "2022-10-09T15:36:25.481Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/ec/0a/1582790232fef6c2aa201f345577306b8bfe465c2c665dec04c86a016879/bcrypt-4.0.1-cp36-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.manylinux_2_24_aarch64.whl", hash = "sha256:08d2947c490093a11416df18043c27abe3921558d2c03e2076ccb28a116cb6d0", size = 583044, upload-time = "2022-10-09T15:37:09.447Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/41/16/49ff5146fb815742ad58cafb5034907aa7f166b1344d0ddd7fd1c818bd17/bcrypt-4.0.1-cp36-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:0eaa47d4661c326bfc9d08d16debbc4edf78778e6aaba29c1bc7ce67214d4410", size = 583189, upload-time = "2022-10-09T15:37:10.69Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/aa/48/fd2b197a9741fa790ba0b88a9b10b5e88e62ff5cf3e1bc96d8354d7ce613/bcrypt-4.0.1-cp36-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ae88eca3024bb34bb3430f964beab71226e761f51b912de5133470b649d82344", size = 593473, upload-time = "2022-10-09T15:36:27.195Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/7d/50/e683d8418974a602ba40899c8a5c38b3decaf5a4d36c32fc65dce454d8a8/bcrypt-4.0.1-cp36-abi3-manylinux_2_24_x86_64.whl", hash = "sha256:a522427293d77e1c29e303fc282e2d71864579527a04ddcfda6d4f8396c6c36a", size = 593249, upload-time = "2022-10-09T15:36:28.481Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/fb/a7/ee4561fd9b78ca23c8e5591c150cc58626a5dfb169345ab18e1c2c664ee0/bcrypt-4.0.1-cp36-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:fbdaec13c5105f0c4e5c52614d04f0bca5f5af007910daa8b6b12095edaa67b3", size = 583586, upload-time = "2022-10-09T15:37:11.962Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/64/fe/da28a5916128d541da0993328dc5cf4b43dfbf6655f2c7a2abe26ca2dc88/bcrypt-4.0.1-cp36-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:ca3204d00d3cb2dfed07f2d74a25f12fc12f73e606fcaa6975d1f7ae69cacbb2", size = 593659, upload-time = "2022-10-09T15:36:30.049Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/dd/4f/3632a69ce344c1551f7c9803196b191a8181c6a1ad2362c225581ef0d383/bcrypt-4.0.1-cp36-abi3-musllinux_1_1_aarch64.whl", hash = "sha256:089098effa1bc35dc055366740a067a2fc76987e8ec75349eb9484061c54f535", size = 613116, upload-time = "2022-10-09T15:37:14.107Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/87/69/edacb37481d360d06fc947dab5734aaf511acb7d1a1f9e2849454376c0f8/bcrypt-4.0.1-cp36-abi3-musllinux_1_1_x86_64.whl", hash = "sha256:e9a51bbfe7e9802b5f3508687758b564069ba937748ad7b9e890086290d2f79e", size = 624290, upload-time = "2022-10-09T15:36:31.251Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/aa/ca/6a534669890725cbb8c1fb4622019be31813c8edaa7b6d5b62fc9360a17e/bcrypt-4.0.1-cp36-abi3-win32.whl", hash = "sha256:2caffdae059e06ac23fce178d31b4a702f2a3264c20bfb5ff541b338194d8fab", size = 159428, upload-time = "2022-10-09T15:36:32.893Z" }, + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/46/81/d8c22cd7e5e1c6a7d48e41a1d1d46c92f17dae70a54d9814f746e6027dec/bcrypt-4.0.1-cp36-abi3-win_amd64.whl", hash = "sha256:8a68f4341daf7522fe8d73874de8906f3a339048ba406be6ddc1b3ccb16fc0d9", size = 152930, upload-time = "2022-10-09T15:36:34.635Z" }, +] + [[package]] name = "beautifulsoup4" version = "4.15.0" @@ -465,9 +488,11 @@ source = { editable = "common" } dependencies = [ { name = "apscheduler" }, { name = "asyncmy" }, + { name = "bcrypt" }, { name = "boto3" }, { name = "fastapi" }, { name = "greenlet" }, + { name = "passlib" }, { name = "pydantic-settings" }, { name = "sqlalchemy" }, ] @@ -476,9 +501,11 @@ dependencies = [ requires-dist = [ { name = "apscheduler", specifier = ">=3.11.3" }, { name = "asyncmy", specifier = "==0.2.11" }, + { name = "bcrypt", specifier = ">=4.0,<4.1" }, { name = "boto3", specifier = ">=1.34,<2" }, { name = "fastapi", specifier = "==0.116.1" }, { name = "greenlet", specifier = ">=3.0.0" }, + { name = "passlib", specifier = "==1.7.4" }, { name = "pydantic-settings", specifier = ">=2.14.2" }, { name = "sqlalchemy", specifier = "==2.0.51" }, ] @@ -1357,6 +1384,15 @@ wheels = [ { url = "https://pypi.tuna.tsinghua.edu.cn/packages/99/5d/8268b644392ee874ee82a635cd0df1773de230bde356c38de28e298392cc/parso-0.8.7-py2.py3-none-any.whl", hash = "sha256:a8926eb2a1b915486941fdbd31e86a4baf88fe8c210f25f2f35ecec5b574ca1c", size = 107025, upload-time = "2026-05-01T23:12:58.867Z" }, ] +[[package]] +name = "passlib" +version = "1.7.4" +source = { registry = "https://pypi.tuna.tsinghua.edu.cn/simple/" } +sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/b6/06/9da9ee59a67fae7761aab3ccc84fa4f3f33f125b370f1ccdb915bf967c11/passlib-1.7.4.tar.gz", hash = "sha256:defd50f72b65c5402ab2c573830a6978e5f202ad0d984793c8dde2c4152ebe04", size = 689844, upload-time = "2020-10-08T19:00:52.121Z" } +wheels = [ + { url = "https://pypi.tuna.tsinghua.edu.cn/packages/3b/a4/ab6b7589382ca3df236e03faa71deac88cae040af60c071a78d254a62172/passlib-1.7.4-py2.py3-none-any.whl", hash = "sha256:aa6bca462b8d8bda89c70b382f0c298a20b5560af6cbfa2dce410c0a2fb669f1", size = 525554, upload-time = "2020-10-08T19:00:49.856Z" }, +] + [[package]] name = "pexpect" version = "4.9.0"