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

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

Browse Source
This commit is contained in:
Winnie
2026-08-05 17:46:05 +08:00
parent 30cde4a56c 1ef9961307
commit 2bfe43a63f
101 changed files with 8797 additions and 3323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/backend/auth.py
+14 -3
View File
@@ -22,6 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from backend.dependencies import database_session
from common.auth.jwt import JwtError, issue_jwt, verify_jwt_token
from common.auth.membership import resolve_is_system_admin
from common.auth.passwords import verify_password
from common.db.models import Roles, Users, WorkspaceMembers, Workspaces
from common.ids import new_ulid
@@ -55,7 +56,12 @@ def _clear_session_cookie(response: Response) -> None:
response.delete_cookie(key=COOKIE_NAME, path="/")
def _user_payload(user: Users, role_code: str | None = None) -> dict[str, Any]:
def _user_payload(
user: Users,
role_code: str | None = None,
*,
is_system_admin: bool = False,
) -> dict[str, Any]:
return {
"user_id": user.user_id,
"username": user.username,
@@ -63,6 +69,7 @@ def _user_payload(user: Users, role_code: str | None = None) -> dict[str, Any]:
"email": user.email,
"status": user.status,
"role_code": role_code,
"is_system_admin": is_system_admin,
}
@@ -161,10 +168,12 @@ async def login(
token = issue_jwt(user.user_id, ttl_seconds=COOKIE_TTL_SECONDS)
_set_session_cookie(request, response, token)
is_system_admin = await resolve_is_system_admin(session, user)
return {
"request_id": new_ulid(),
"data": {
"user": _user_payload(user, user_role_code),
"user": _user_payload(user, user_role_code, is_system_admin=is_system_admin),
"workspaces": workspaces,
"default_workspace_id": default_workspace_id,
},
@@ -238,10 +247,12 @@ async def me(
if user_role_code is None and rows:
user_role_code = rows[0][1].role_code
is_system_admin = await resolve_is_system_admin(session, user)
return {
"request_id": new_ulid(),
"data": {
"user": _user_payload(user, user_role_code),
"user": _user_payload(user, user_role_code, is_system_admin=is_system_admin),
"workspaces": workspaces,
"default_workspace_id": default_workspace_id,
},
backend/src/backend/dependencies.py
+63 -12
View File
@@ -31,10 +31,15 @@ from dataclasses import dataclass
from typing import AsyncIterator
from fastapi import Depends, HTTPException, Query, Request, status
from sqlalchemy import select
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.auth.membership import (
MembershipError,
load_active_membership,
resolve_is_system_admin,
)
from common.db import session_scope
from common.db.models import Roles, Users, Workspaces
from common.ids import new_ulid
@@ -49,10 +54,18 @@ class RequestContext:
user: Users
workspace: Workspaces
role: Roles
# True when the requester holds the platform-scoped admin role (via
# Users.platform_role_id). Same flag exposed on ``/api/v1/auth/me``
# so the frontend can render the platform-admin entry point. System
# admins have full control over every workspace — including disabled
# ones — so ``request_context`` lets them through and this flag is
# the single signal handlers use to gate platform-only operations.
is_system_admin: bool = False
@property
def is_admin(self) -> bool:
return self.role.role_code == "admin"
"""Workspace admin OR system admin (the latter is strictly stronger)."""
return self.is_system_admin or self.role.role_code == "admin"
async def database_session(request: Request) -> AsyncIterator[AsyncSession]:
@@ -97,7 +110,7 @@ async def request_context(
),
session: AsyncSession = Depends(database_session),
) -> RequestContext:
"""Verify JWT and load the user's active membership for ``workspace_id``.
"""Verify JWT and load the user's 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
@@ -105,21 +118,59 @@ async def request_context(
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.
System admins (Users.platform_role_id → admin role) bypass the
active-membership requirement: they can address disabled workspaces
because they own the platform. Non-admin users still need an
active ``WorkspaceMembers`` row in an active ``Workspaces`` row.
"""
user = await current_user(request, session)
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
is_system_admin = await resolve_is_system_admin(session, user)
if is_system_admin:
# System admin: any workspace (active or disabled) is fine.
# Still 404 if the workspace_id is genuinely unknown — the
# ``?workspace_id=`` query param is part of the URL contract.
workspace = await session.get(Workspaces, workspace_id)
if workspace is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"workspace not found",
)
# Synthesize a role object so the rest of ``RequestContext``
# (and downstream ``is_admin`` checks) keep working without
# branching on whether a real WorkspaceMembers row exists.
role = await _load_admin_role(session)
if role is None:
# The seed migration creates this row, so this is a
# hard config error if it's missing.
raise HTTPException(
status.HTTP_500_INTERNAL_SERVER_ERROR,
"admin role not configured",
)
else:
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,
is_system_admin=is_system_admin,
)
async def _load_admin_role(session: AsyncSession) -> Roles | None:
"""Return the singleton ``role_code='admin'`` row (None if absent)."""
return await session.scalar(
select(Roles).where(Roles.role_code == "admin")
)
backend/src/backend/main.py
+20 -29
View File
@@ -10,9 +10,16 @@ from fastapi.routing import APIRoute
from common.config import settings
from common.db import create_database_engine, create_session_factory
from common.service_app import create_service_app
from common.storage import RustFSObjectStore
from common.storage import (
AsyncStorageBackend,
PURPOSE_BUCKETS,
actual_bucket_name,
build_storage_config,
create_storage,
)
from backend.admin import router as admin_router
from backend.auth import router as auth_router
from backend.platform import router as platform_router
from backend.jupyter import router as jupyter_router
from backend.resources import router as resources_router
from backend.runtime_client import RuntimeClient
@@ -21,7 +28,6 @@ from backend.schedule_runs import router as schedule_runs_router
from backend.schedules import router as schedules_router
from backend.scripts import router as scripts_router
from backend.storage_api import app as storage_app
from backend.storage_client import StorageClient
@asynccontextmanager
@@ -29,32 +35,17 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
engine = create_database_engine(settings.database_url)
app.state.session_factory = create_session_factory(engine)
# Storage API is now part of the backend process. Platform routers keep
# their existing client contract, but calls are dispatched in-process.
app.state.object_store = RustFSObjectStore(
internal_endpoint=settings.rustfs_endpoint,
access_key=settings.rustfs_access_key,
secret_key=settings.rustfs_secret_key,
)
# 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,
bucket,
)
app.state.default_bucket = settings.rustfs_workspace_bucket
storage_http_client = httpx.AsyncClient(
transport=httpx.ASGITransport(app=app),
base_url="http://backend.internal",
timeout=httpx.Timeout(30.0),
)
app.state.storage_client = StorageClient(storage_http_client)
# Storage API is part of the backend process. Platform routers call
# the helpers in ``backend.services.storage`` directly (in-process),
# so no HTTP client is needed. The dict is keyed by the actual
# bucket name (e.g. "versions"), matching ``UploadSessions.bucket_name``
# and ``StorageObjects.bucket_name`` so call sites can do
# ``object_stores[upload.bucket_name].put(...)`` directly.
app.state.object_stores: dict[str, AsyncStorageBackend] = {
actual_bucket_name(purpose): create_storage(build_storage_config(purpose))
for purpose in PURPOSE_BUCKETS
}
app.state.default_bucket = settings.s3_workspace_bucket
runtime_http_client = httpx.AsyncClient(
base_url=settings.runtime_api_url,
timeout=httpx.Timeout(30.0),
@@ -71,7 +62,6 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
finally:
await rclone_http_client.aclose()
await runtime_http_client.aclose()
await storage_http_client.aclose()
await engine.dispose()
@@ -86,6 +76,7 @@ app.include_router(schedule_runs_router)
app.include_router(schedules_router)
app.include_router(scripts_router)
app.include_router(admin_router)
app.include_router(platform_router)
# Reuse the proven storage endpoints without running another FastAPI service.
for route in storage_app.routes:
backend/src/backend/platform.py
+607
View File
@@ -0,0 +1,607 @@
"""System-admin (platform-scope) endpoints for workspace & membership management.
All routes under ``/api/v1/platform/*`` are gated by
:func:`system_admin_context`, which requires the requester to hold a
``Users.platform_role_id`` pointing to a ``Roles`` row whose
``role_code == 'admin'``. Unlike ``backend.dependencies.request_context``,
this dependency does NOT require an active workspace membership — system
admins can manage workspaces before/without being a member of any.
Endpoints
---------
Workspace CRUD::
GET /workspaces — list non-deleted workspaces
POST /workspaces — create a new workspace
GET /workspaces/{workspace_id} — single workspace (incl. disabled)
PATCH /workspaces/{workspace_id} — update editable fields
DELETE /workspaces/{workspace_id} — soft delete (cascades memberships)
Workspace membership CRUD::
GET /workspaces/{workspace_id}/members — list active members
POST /workspaces/{workspace_id}/members — add a member
PATCH /workspaces/{workspace_id}/members/{user_id} — update role/status
DELETE /workspaces/{workspace_id}/members/{user_id} — remove a member
Invariants
----------
* Every workspace must always retain at least one active ``admin`` member.
* A system admin cannot remove their own workspace membership via
``DELETE .../members/{self}``; the only escape is to delete the entire
workspace, which cascades membership soft-deletion.
* ``DELETE /workspaces/{id}`` is allowed from any non-disabled status and
sets ``status='disabled'`` + ``is_deleted=1`` + ``deleted_at`` on the
workspace and every one of its active memberships.
"""
from __future__ import annotations
import datetime
import re
from dataclasses import dataclass
from typing import Any, Literal
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from backend.dependencies import current_user, database_session
from common.db.models import Roles, Users, WorkspaceMembers, Workspaces
from common.ids import new_ulid
router = APIRouter(prefix="/api/v1/platform", tags=["platform"])
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
WORKSPACE_CODE_PATTERN = re.compile(r"^[a-z0-9-]{3,32}$")
LIST_PAGE_SIZE = 100
WORKSPACE_EDITABLE_STATUS = ("active", "archived")
MEMBER_ROLE_CODES = ("admin", "developer")
MEMBER_STATUS_VALUES = ("active", "disabled", "locked")
# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------
class WorkspaceCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
workspace_code: str = Field(min_length=3, max_length=32)
workspace_name: str = Field(min_length=1, max_length=150)
quota_bytes: int = Field(default=0, ge=0)
description: str | None = Field(default=None, max_length=1000)
class WorkspaceUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
workspace_name: str | None = Field(default=None, min_length=1, max_length=150)
quota_bytes: int | None = Field(default=None, ge=0)
description: str | None = Field(default=None, max_length=1000)
# 'disabled' is rejected here on purpose — soft delete must go through DELETE.
status: Literal["active", "archived"] | None = None
class MemberCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
user_id: str = Field(min_length=26, max_length=26)
role_code: Literal["admin", "developer"]
class MemberUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
role_code: Literal["admin", "developer"] | None = None
member_status: Literal["active", "disabled", "locked"] | None = None
# ---------------------------------------------------------------------------
# System-admin context dependency
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class SystemAdminContext:
"""Resolved identity for a system-admin request.
Carries the request id, the authenticated user row, and the resolved
``Roles`` row the user holds via ``Users.platform_role_id``. By
construction the role's ``role_code`` is ``"admin"``.
"""
request_id: str
user: Users
platform_role: Roles
async def system_admin_context(
request: Request,
session: AsyncSession = Depends(database_session),
) -> SystemAdminContext:
"""Resolve the requester as a system admin.
Steps:
1. Reuse :func:`backend.dependencies.current_user` to validate the JWT
cookie and fetch the active ``Users`` row (raises 401 on failure).
2. Require ``Users.platform_role_id`` to point to a row whose
``role_code == 'admin'`` — anything else is 403.
"""
user = await current_user(request, session)
if user.platform_role_id is None:
raise HTTPException(
status.HTTP_403_FORBIDDEN,
"需要系统管理员权限",
)
platform_role = await session.scalar(
select(Roles).where(Roles.role_id == user.platform_role_id)
)
if platform_role is None or platform_role.role_code != "admin":
raise HTTPException(
status.HTTP_403_FORBIDDEN,
"需要系统管理员权限",
)
request_id = request.headers.get("X-Request-ID") or new_ulid()
return SystemAdminContext(
request_id=request_id,
user=user,
platform_role=platform_role,
)
# ---------------------------------------------------------------------------
# Payload helpers
# ---------------------------------------------------------------------------
def workspace_payload(workspace: Workspaces) -> dict[str, Any]:
return {
"workspace_id": workspace.workspace_id,
"workspace_code": workspace.workspace_code,
"workspace_name": workspace.workspace_name,
"active_root_uri": workspace.active_root_uri,
"quota_bytes": workspace.quota_bytes,
"status": workspace.status,
"description": workspace.description,
"created_by": workspace.created_by,
"created_at": workspace.created_at.isoformat(),
"updated_at": (
workspace.updated_at.isoformat() if workspace.updated_at else None
),
}
def member_payload(
user: Users,
role: Roles,
membership: WorkspaceMembers,
) -> dict[str, Any]:
return {
"user_id": user.user_id,
"username": user.username,
"display_name": user.display_name,
"email": user.email,
"user_status": user.status,
"role_code": role.role_code,
"role_name": role.role_name,
"member_status": membership.member_status,
"joined_at": membership.joined_at.isoformat(),
}
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
async def _load_workspace(session: AsyncSession, workspace_id: str) -> Workspaces:
workspace = await session.get(Workspaces, workspace_id)
if workspace is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "workspace 不存在")
return workspace
async def _load_role_by_code(session: AsyncSession, role_code: str) -> Roles:
role = await session.scalar(select(Roles).where(Roles.role_code == role_code))
if role is None:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
f"角色 {role_code} 不存在",
)
return role
async def _count_active_admins(
session: AsyncSession,
workspace_id: str,
exclude_user_id: str | None = None,
) -> int:
"""Count active admin members of ``workspace_id``.
Pass ``exclude_user_id`` when checking "would X be the last admin?"
before mutating X.
"""
admin_role = await _load_role_by_code(session, "admin")
stmt = (
select(func.count())
.select_from(WorkspaceMembers)
.where(
WorkspaceMembers.workspace_id == workspace_id,
WorkspaceMembers.role_id == admin_role.role_id,
WorkspaceMembers.member_status == "active",
WorkspaceMembers.is_deleted == 0,
)
)
if exclude_user_id is not None:
stmt = stmt.where(WorkspaceMembers.user_id != exclude_user_id)
return int(await session.scalar(stmt) or 0)
def _envelope(request_id: str, data: Any, meta: dict[str, Any] | None = None) -> dict[str, Any]:
return {
"request_id": request_id,
"data": data,
"meta": meta or {},
}
# ---------------------------------------------------------------------------
# Workspace CRUD
# ---------------------------------------------------------------------------
@router.get("/workspaces")
async def list_workspaces(
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""List active/archived workspaces. Soft-deleted rows are filtered out.
Silent ``pageSize=100`` cap — YAGNI on real pagination until needed.
"""
rows = (
await session.execute(
select(Workspaces)
.where(
Workspaces.status != "disabled",
Workspaces.is_deleted == 0,
)
.order_by(Workspaces.created_at, Workspaces.workspace_id)
.limit(LIST_PAGE_SIZE)
)
).scalars().all()
return _envelope(
context.request_id,
[workspace_payload(w) for w in rows],
{"count": len(rows), "page_size": LIST_PAGE_SIZE},
)
@router.post("/workspaces", status_code=status.HTTP_201_CREATED)
async def create_workspace(
payload: WorkspaceCreate,
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Create a workspace and auto-join the creator as an admin member."""
if not WORKSPACE_CODE_PATTERN.fullmatch(payload.workspace_code):
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
"workspace_code 必须匹配 ^[a-z0-9-]{3,32}$",
)
duplicate = await session.scalar(
select(Workspaces.workspace_id).where(
Workspaces.workspace_code == payload.workspace_code,
)
)
if duplicate is not None:
raise HTTPException(status.HTTP_409_CONFLICT, "workspace_code 已存在")
admin_role = await _load_role_by_code(session, "admin")
workspace_id = new_ulid()
workspace = Workspaces(
workspace_id=workspace_id,
workspace_code=payload.workspace_code,
workspace_name=payload.workspace_name,
active_root_uri=f"s3://workspaces/{workspace_id}/",
quota_bytes=payload.quota_bytes,
status="active",
created_by=context.user.user_id,
description=payload.description,
)
session.add(workspace)
session.add(
WorkspaceMembers(
workspace_id=workspace_id,
user_id=context.user.user_id,
role_id=admin_role.role_id,
member_status="active",
)
)
await session.flush()
await session.refresh(workspace)
return _envelope(context.request_id, workspace_payload(workspace))
@router.get("/workspaces/{workspace_id}")
async def get_workspace(
workspace_id: str,
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Fetch a single workspace — even soft-deleted ones are reachable."""
workspace = await _load_workspace(session, workspace_id)
return _envelope(context.request_id, workspace_payload(workspace))
@router.patch("/workspaces/{workspace_id}")
async def update_workspace(
workspace_id: str,
payload: WorkspaceUpdate,
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Patch editable workspace fields. ``status='disabled'`` is rejected."""
workspace = await _load_workspace(session, workspace_id)
if workspace.status == "disabled":
raise HTTPException(
status.HTTP_409_CONFLICT,
"workspace 已删除,无法修改",
)
if payload.workspace_name is not None:
workspace.workspace_name = payload.workspace_name.strip()
if payload.quota_bytes is not None:
workspace.quota_bytes = payload.quota_bytes
if payload.description is not None:
workspace.description = payload.description
if payload.status is not None:
workspace.status = payload.status
await session.flush()
await session.refresh(workspace)
return _envelope(context.request_id, workspace_payload(workspace))
@router.delete("/workspaces/{workspace_id}")
async def delete_workspace(
workspace_id: str,
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Soft-delete a workspace and cascade-soft-delete its memberships.
Allowed from any non-disabled status (active or archived). The
membership cascade is what lets system admins leave a workspace —
there is no per-member DELETE escape for self-removal.
"""
workspace = await _load_workspace(session, workspace_id)
if workspace.status == "disabled":
raise HTTPException(
status.HTTP_409_CONFLICT,
"workspace 已被删除",
)
now = datetime.datetime.utcnow()
workspace.status = "disabled"
workspace.is_deleted = 1
workspace.deleted_at = now
await session.execute(
update(WorkspaceMembers)
.where(
WorkspaceMembers.workspace_id == workspace_id,
WorkspaceMembers.is_deleted == 0,
)
.values(is_deleted=1, deleted_at=now)
)
await session.flush()
await session.refresh(workspace)
return _envelope(context.request_id, workspace_payload(workspace))
# ---------------------------------------------------------------------------
# Workspace membership CRUD
# ---------------------------------------------------------------------------
@router.get("/workspaces/{workspace_id}/members")
async def list_members(
workspace_id: str,
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""List active and historical (non-soft-deleted) members of a workspace."""
await _load_workspace(session, workspace_id)
rows = (
await session.execute(
select(Users, Roles, WorkspaceMembers)
.join(
WorkspaceMembers,
WorkspaceMembers.user_id == Users.user_id,
)
.join(Roles, Roles.role_id == WorkspaceMembers.role_id)
.where(
WorkspaceMembers.workspace_id == workspace_id,
WorkspaceMembers.is_deleted == 0,
)
.order_by(WorkspaceMembers.joined_at, Users.user_id)
.limit(LIST_PAGE_SIZE)
)
).all()
return _envelope(
context.request_id,
[member_payload(u, r, m) for u, r, m in rows],
{"count": len(rows), "page_size": LIST_PAGE_SIZE},
)
@router.post(
"/workspaces/{workspace_id}/members",
status_code=status.HTTP_201_CREATED,
)
async def add_member(
workspace_id: str,
payload: MemberCreate,
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Add a user to a workspace. The new row starts with member_status='active'."""
await _load_workspace(session, workspace_id)
user = await session.get(Users, payload.user_id)
if user is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "用户不存在")
role = await _load_role_by_code(session, payload.role_code)
duplicate = await session.scalar(
select(WorkspaceMembers.user_id).where(
WorkspaceMembers.workspace_id == workspace_id,
WorkspaceMembers.user_id == payload.user_id,
WorkspaceMembers.is_deleted == 0,
)
)
if duplicate is not None:
raise HTTPException(
status.HTTP_409_CONFLICT,
"用户已是该 workspace 成员",
)
membership = WorkspaceMembers(
workspace_id=workspace_id,
user_id=payload.user_id,
role_id=role.role_id,
member_status="active",
)
session.add(membership)
await session.flush()
await session.refresh(membership)
return _envelope(context.request_id, member_payload(user, role, membership))
@router.patch("/workspaces/{workspace_id}/members/{user_id}")
async def update_member(
workspace_id: str,
user_id: str,
payload: MemberUpdate,
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Update a member's role and/or status. Last-admin guard applies."""
await _load_workspace(session, workspace_id)
row = (
await session.execute(
select(Users, Roles, WorkspaceMembers)
.join(
WorkspaceMembers,
WorkspaceMembers.user_id == Users.user_id,
)
.join(Roles, Roles.role_id == WorkspaceMembers.role_id)
.where(
WorkspaceMembers.workspace_id == workspace_id,
WorkspaceMembers.user_id == user_id,
WorkspaceMembers.is_deleted == 0,
)
)
).first()
if row is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "成员不存在")
user, role, membership = row
next_role = role
if payload.role_code is not None and payload.role_code != role.role_code:
if (
role.role_code == "admin"
and payload.role_code != "admin"
and membership.member_status == "active"
):
remaining = await _count_active_admins(
session, workspace_id, exclude_user_id=user_id,
)
if remaining == 0:
raise HTTPException(
status.HTTP_409_CONFLICT,
"workspace 必须保留至少一个 admin",
)
next_role = await _load_role_by_code(session, payload.role_code)
membership.role_id = next_role.role_id
if payload.member_status is not None and payload.member_status != membership.member_status:
if (
role.role_code == "admin"
and payload.member_status != "active"
):
remaining = await _count_active_admins(
session, workspace_id, exclude_user_id=user_id,
)
if remaining == 0:
raise HTTPException(
status.HTTP_409_CONFLICT,
"workspace 必须保留至少一个 admin",
)
membership.member_status = payload.member_status
await session.flush()
await session.refresh(membership)
return _envelope(context.request_id, member_payload(user, next_role, membership))
@router.delete("/workspaces/{workspace_id}/members/{user_id}")
async def remove_member(
workspace_id: str,
user_id: str,
context: SystemAdminContext = Depends(system_admin_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Soft-delete a workspace membership.
System admins cannot remove themselves — the only escape is to delete
the entire workspace, which cascades membership soft-deletion.
"""
await _load_workspace(session, workspace_id)
if user_id == context.user.user_id:
raise HTTPException(
status.HTTP_403_FORBIDDEN,
"系统管理员不能把自己从 workspace 移除;如需退出,请删除整个 workspace",
)
row = (
await session.execute(
select(Roles, WorkspaceMembers)
.join(Roles, Roles.role_id == WorkspaceMembers.role_id)
.where(
WorkspaceMembers.workspace_id == workspace_id,
WorkspaceMembers.user_id == user_id,
WorkspaceMembers.is_deleted == 0,
)
)
).first()
if row is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "成员不存在")
role, membership = row
if role.role_code == "admin" and membership.member_status == "active":
remaining = await _count_active_admins(
session, workspace_id, exclude_user_id=user_id,
)
if remaining == 0:
raise HTTPException(
status.HTTP_409_CONFLICT,
"workspace 必须保留至少一个 admin",
)
membership.is_deleted = 1
membership.deleted_at = datetime.datetime.utcnow()
await session.flush()
return _envelope(
context.request_id,
{"workspace_id": workspace_id, "user_id": user_id, "removed": True},
)
__all__ = [
"router",
"SystemAdminContext",
"system_admin_context",
]
backend/src/backend/resources.py
+89 -43
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import base64
from datetime import UTC, datetime
from typing import Any
@@ -9,6 +10,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
from common.db.models import DataResources, StorageObjects
from common.ids import new_ulid
from common.storage.schemas import (
CreateUploadRequest,
DownloadUrlRequest,
ServerObjectRequest,
)
from backend.dependencies import (
RequestContext,
database_session,
@@ -19,6 +25,13 @@ from backend.schemas import (
CreateResourceUploadRequest,
DownloadUrlRequest,
)
from backend.services.storage import (
create_download_url_payload,
create_server_object_payload,
create_upload_record,
soft_delete_object,
upload_bytes_to_session,
)
router = APIRouter(prefix="/api/v1/data-resources", tags=["data-resources"])
@@ -68,52 +81,95 @@ async def create_resource_upload(
alias="Idempotency-Key",
),
) -> dict[str, Any]:
data = await request.app.state.storage_client.create_upload(
{
"workspace_id": context.workspace.workspace_id,
"user_id": context.user.user_id,
"usage_type": "data_resource",
"file_name": payload.file_name,
"content_type": payload.content_type,
"expected_size_bytes": payload.expected_size_bytes,
"expected_hash": payload.expected_hash,
"idempotency_key": idempotency_key,
}
data = await create_upload_record(
CreateUploadRequest(
workspace_id=context.workspace.workspace_id,
user_id=context.user.user_id,
usage_type="data_resource",
file_name=payload.file_name,
content_type=payload.content_type,
expected_size_bytes=payload.expected_size_bytes,
expected_hash=payload.expected_hash,
idempotency_key=idempotency_key,
),
session,
request,
)
return {"request_id": context.request_id, "data": data, "meta": {}}
@router.post("/uploads/{upload_id}/complete")
async def complete_resource_upload(
@router.put("/uploads/{upload_id}")
async def upload_resource_bytes(
upload_id: str,
request: Request,
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Server-proxied upload step 2: PUT the raw bytes here.
Replaces the old 3-step presign-PUT flow. The new flow is:
POST /uploads → {upload_id, upload_path, ...}
PUT /uploads/{upload_id} ← this route
(3) The frontend then calls a separate bind route to attach the
resulting StorageObjects row to a DataResources row.
"""
item = await upload_bytes_to_session(upload_id, session, request)
return {
"request_id": context.request_id,
"data": {"storage_object_id": item.storage_object_id},
"meta": {},
}
@router.post("/uploads/{upload_id}/bind")
async def bind_resource(
upload_id: str,
payload: CompleteResourceUploadRequest,
request: Request,
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
storage_data = await request.app.state.storage_client.complete_upload(
upload_id,
{
"usage_type": "data_resource",
"file_name": payload.resource_name,
"visibility": payload.visibility,
"is_immutable": False,
},
"""Bind a completed upload to a DataResources row.
Caller must have already PUT the bytes (see ``PUT /uploads/{id}``).
This route attaches the resource_name / description / visibility to
the StorageObjects row + creates the DataResources row that points
to it.
"""
from common.db.models import UploadSessions
upload = await session.scalar(
select(UploadSessions).where(UploadSessions.upload_id == upload_id)
)
if storage_data["workspace_id"] != context.workspace.workspace_id:
if upload is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found")
if upload.storage_object_id is None:
raise HTTPException(
status.HTTP_409_CONFLICT,
"upload has no completed object; PUT the bytes first",
)
if upload.workspace_id != context.workspace.workspace_id:
raise HTTPException(
status.HTTP_403_FORBIDDEN,
"upload belongs to another workspace",
)
if storage_data["owner_user_id"] != context.user.user_id:
if upload.user_id != context.user.user_id:
raise HTTPException(
status.HTTP_403_FORBIDDEN,
"upload belongs to another user",
)
item = await session.get(StorageObjects, upload.storage_object_id)
if item is None:
raise HTTPException(
status.HTTP_409_CONFLICT,
"storage object metadata is missing",
)
# Persist resource_name / description / visibility override.
item.visibility = payload.visibility
# (description lives on DataResources, not on StorageObjects.)
existing = await session.scalar(
select(DataResources).where(
DataResources.storage_object_id
== storage_data["storage_object_id"]
DataResources.storage_object_id == item.storage_object_id
)
)
reused = existing is not None
@@ -121,7 +177,7 @@ async def complete_resource_upload(
existing = DataResources(
resource_id=new_ulid(),
workspace_id=context.workspace.workspace_id,
storage_object_id=storage_data["storage_object_id"],
storage_object_id=item.storage_object_id,
owner_user_id=context.user.user_id,
resource_name=payload.resource_name,
description=payload.description,
@@ -131,18 +187,9 @@ async def complete_resource_upload(
session.add(existing)
await session.flush()
await session.refresh(existing)
storage_object = await session.get(
StorageObjects,
existing.storage_object_id,
)
if storage_object is None:
raise HTTPException(
status.HTTP_409_CONFLICT,
"storage object metadata is missing",
)
return {
"request_id": context.request_id,
"data": resource_payload(existing, storage_object),
"data": resource_payload(existing, item),
"meta": {"reused": reused},
}
@@ -254,11 +301,12 @@ async def resource_download_url(
context,
session,
)
data = await request.app.state.storage_client.create_download_url(
resource.storage_object_id,
payload.expires_seconds,
data = await create_download_url_payload(
await session.get(StorageObjects, resource.storage_object_id),
DownloadUrlRequest(expires_seconds=payload.expires_seconds),
request,
)
return {"request_id": context.request_id, "data": data, "meta": {}}
return {"request_id": context.request_id, "data": data["data"], "meta": {}}
@router.delete("/{resource_id}")
@@ -281,9 +329,7 @@ async def delete_resource(
status.HTTP_403_FORBIDDEN,
"resource can only be deleted by its owner or an administrator",
)
await request.app.state.storage_client.delete_object(
resource.storage_object_id
)
await soft_delete_object(resource.storage_object_id, request, session)
resource.status = "deleted"
resource.deleted_at = datetime.now(UTC).replace(tzinfo=None)
return {
backend/src/backend/scripts.py
+33 -21
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import base64
import hashlib
import json
import mimetypes
@@ -34,6 +35,11 @@ from backend.dependencies import (
request_context,
)
from backend.runtime_client import RuntimeClientError
from backend.services.storage import (
create_download_url_payload,
create_server_object_payload,
)
from common.storage.schemas import ServerObjectRequest
from backend.schemas import (
CreateScriptRequest,
CreateWorkspaceDirectoryRequest,
@@ -359,13 +365,13 @@ async def create_script_record(
# Build a real StorageObjects row so the file participates in
# workspace-tree / list / get queries that JOIN this table. The
# bytes live in the Jupyter mount; rclone replicates them to
# RustFS asynchronously. We mark the row "available" because the
# bytes live in the Jupyter mount; rclone replicates them to S3
# asynchronously. We mark the row "available" because the
# file is queryable as a workspace file from the user's POV; the
# storage_uri points at where the replicated bytes will land.
object_id = new_ulid()
object_key = f"{workspace_id}/{jupyter_name}"
bucket_name = settings.rustfs_workspace_bucket
bucket_name = settings.s3_workspace_bucket
relative_path = user_relative_path(context, jupyter_name)
mime_type = mimetypes.guess_type(jupyter_name)[0]
storage_object = StorageObjects(
@@ -374,7 +380,7 @@ async def create_script_record(
owner_user_id=context.user.user_id,
object_type="file",
usage_type="working_copy",
storage_backend="rustfs",
storage_backend="s3",
bucket_name=bucket_name,
object_key=object_key,
object_key_hash=hashlib.sha256(object_key.encode("utf-8")).digest(),
@@ -587,7 +593,7 @@ async def create_workspace_directory(
status.HTTP_404_NOT_FOUND,
"parent directory not found",
)
# RustFS has no real directory objects — the prefix is implicitly
# S3 has no real directory objects — the prefix is implicitly
# created when a file is uploaded. Conflict detection is best-effort.
existing = await session.scalar(
select(StorageObjects.storage_object_id).where(
@@ -931,33 +937,38 @@ async def publish_version(
mimetypes.guess_type(script.script_name)[0]
or "application/octet-stream"
)
artifact = await request.app.state.storage_client.create_server_object(
workspace_id=context.workspace.workspace_id,
user_id=context.user.user_id,
usage_type="version_artifact",
file_name=script.script_name,
content_type=content_type,
content=content,
visibility=payload.visibility,
is_immutable=True,
idempotency_key=f"version:{script.script_id}:{content_hash}",
artifact = await create_server_object_payload(
ServerObjectRequest(
workspace_id=context.workspace.workspace_id,
user_id=context.user.user_id,
usage_type="version_artifact",
file_name=script.script_name,
content_type=content_type,
content_base64=base64.b64encode(content).decode("ascii"),
visibility=payload.visibility,
is_immutable=True,
idempotency_key=f"version:{script.script_id}:{content_hash}",
),
request,
session,
)
current_max = await session.scalar(
select(func.max(Versions.version_no)).where(
Versions.script_id == script.script_id
)
)
artifact_data = artifact["data"]
version_no = int(current_max or 0) + 1
version = Versions(
versions_id=new_ulid(),
workspace_id=context.workspace.workspace_id,
script_id=script.script_id,
source_object_id=script.current_object_id,
artifact_object_id=artifact["storage_object_id"],
artifact_object_id=artifact_data["storage_object_id"],
version_no=version_no,
version_label=f"v{version_no}.0",
source_path=jupyter_name,
artifact_path=artifact["storage_uri"],
artifact_path=artifact_data["storage_uri"],
content_hash=content_hash,
file_size_bytes=len(content),
visibility=payload.visibility,
@@ -1123,8 +1134,9 @@ async def version_download_url(
or version.workspace_id != context.workspace.workspace_id
):
raise HTTPException(status.HTTP_404_NOT_FOUND, "version not found")
data = await request.app.state.storage_client.create_download_url(
version.artifact_object_id,
payload.expires_seconds,
data = await create_download_url_payload(
await session.get(StorageObjects, version.artifact_object_id),
DownloadUrlRequest(expires_seconds=payload.expires_seconds),
request,
)
return {"request_id": context.request_id, "data": data, "meta": {}}
return {"request_id": context.request_id, "data": data["data"], "meta": {}}
backend/src/backend/services/__init__.py
View File
backend/src/backend/services/storage.py
+512
View File
@@ -0,0 +1,512 @@
"""In-process storage helpers.
The HTTP ``/internal/v1/*`` routes in ``backend.storage_api`` are wrappers
around these. Other backend modules (``scripts``, ``resources``) and the
schedule worker call these helpers directly instead of going through an
HTTP client — the storage layer lives in the same process, so the
indirection is pointless.
Functions:
create_upload_record — open a new upload session, returning
the upload_path (PUT-bytes) + session row.
upload_bytes_to_session — read raw bytes from request, validate,
call AsyncStorageBackend.put, build
StorageObjects row.
create_server_object_payload — server-side single-call upload (bytes
in JSON via base64). Used for small
artifacts (≤100 KiB).
create_download_url_payload — build a presigned GET URL for one
StorageObjects row.
soft_delete_object — copy-to-trash + delete source + flip row
to "deleted" with deleted_at stamp.
These helpers raise ``HTTPException`` directly because they share an
HTTP-shaped error contract with the routes; callers can let the
exception propagate.
"""
from __future__ import annotations
import base64
import binascii
import hashlib
from datetime import timedelta
from pathlib import PurePosixPath
from typing import Any
from fastapi import HTTPException, Request, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from common.config import settings
from common.db.models import StorageObjects, UploadSessions
from common.ids import new_ulid
from common.storage.schemas import (
CreateUploadRequest,
DownloadUrlRequest,
ServerObjectRequest,
)
# ── shared low-level helpers (module-private) ────────────────────────────
def _safe_file_name(value: str) -> str:
return value.strip() or "upload.bin"
def _utcnow_naive() -> Any:
from datetime import datetime, UTC
return datetime.now(UTC).replace(tzinfo=None)
def _hash_bytes(value: str) -> bytes:
import hashlib as _h
return _h.sha256(value.encode("utf-8")).digest()
def _build_storage_object(
*,
upload: UploadSessions,
file_name: str,
content_type: str,
size_bytes: int,
content_hash: str | None,
visibility: str,
is_immutable: bool,
usage_type: str,
owner_user_id: str | None = None,
) -> StorageObjects:
"""Build the StorageObjects row that pairs with a completed UploadSessions row."""
safe_name = _safe_file_name(file_name)
return StorageObjects(
storage_object_id=new_ulid(),
workspace_id=upload.workspace_id,
owner_user_id=owner_user_id or upload.user_id,
object_type="file",
usage_type=usage_type,
storage_backend="s3",
bucket_name=upload.bucket_name,
object_key=upload.object_key,
object_key_hash=upload.object_key_hash,
storage_uri=f"s3://{upload.bucket_name}/{upload.object_key}",
file_name=safe_name,
file_extension=PurePosixPath(safe_name).suffix.lower() or None,
mime_type=content_type,
size_bytes=size_bytes,
content_hash=content_hash,
object_etag=None,
visibility=visibility,
is_immutable=int(is_immutable),
object_status="available",
created_by=upload.user_id,
)
# ── create_upload_record ────────────────────────────────────────────────
def _resolve_bucket_for_usage(
usage_type: str,
*,
workspace_artifact_bucket: str | None,
) -> str:
"""Mirror of storage_api.resolve_bucket, but pure (no DB / Request)."""
from backend.storage_api import BUCKET_FOR_USAGE
if workspace_artifact_bucket:
return workspace_artifact_bucket
return BUCKET_FOR_USAGE.get(usage_type, settings.s3_workspace_bucket)
async def create_upload_record(
payload: CreateUploadRequest,
session: AsyncSession,
request: Request,
) -> dict[str, Any]:
"""Create or reuse an UploadSessions row.
Returns ``{upload_id, status, upload_path, expires_at}`` for a fresh
session; or ``{upload_id, status: "completed", storage_object: {...}}``
when the idempotency key hits an already-completed upload.
"""
from backend.storage_api import (
require_workspace_member,
normalized_idempotency_key,
BUCKET_FOR_USAGE,
)
workspace = await require_workspace_member(
session, payload.workspace_id, payload.user_id
)
stored_key = normalized_idempotency_key(
payload.workspace_id, payload.user_id, payload.idempotency_key
)
existing = await session.scalar(
select(UploadSessions).where(UploadSessions.idempotency_key == stored_key)
)
if existing is not None:
if (
existing.workspace_id != payload.workspace_id
or existing.user_id != payload.user_id
or existing.expected_size_bytes != payload.expected_size_bytes
or existing.expected_hash != payload.expected_hash
or existing.content_type != payload.content_type
):
raise HTTPException(
status.HTTP_409_CONFLICT,
"idempotency key was used with different upload metadata",
)
upload = existing
else:
from datetime import timedelta
from backend.storage_api import utcnow
bucket_name = _resolve_bucket_for_usage(
payload.usage_type,
workspace_artifact_bucket=workspace.artifact_bucket,
)
# Keep the opaque upload id while preserving the original extension.
# Jupyter selects its editor from this suffix, so an extensionless
# object would make notebooks look like generic JSON/text files.
file_extension = PurePosixPath(_safe_file_name(payload.file_name)).suffix.lower()
object_key = f"{payload.workspace_id}/{new_ulid()}{file_extension}"
upload = UploadSessions(
upload_id=new_ulid(),
workspace_id=payload.workspace_id,
user_id=payload.user_id,
idempotency_key=stored_key,
bucket_name=bucket_name,
object_key=object_key,
object_key_hash=_hash_bytes(object_key),
upload_status="created",
expires_at=utcnow() + timedelta(minutes=15),
expected_size_bytes=payload.expected_size_bytes,
expected_hash=payload.expected_hash,
content_type=payload.content_type,
file_name=payload.file_name,
usage_type=payload.usage_type,
visibility=payload.visibility,
is_immutable=int(payload.is_immutable),
)
session.add(upload)
await session.flush()
if upload.upload_status == "completed" and upload.storage_object_id:
from backend.storage_api import storage_payload
storage_object = await session.get(StorageObjects, upload.storage_object_id)
if storage_object is None or storage_object.object_status != "available":
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,
f"upload cannot continue from status {upload.upload_status}",
)
return {
"upload_id": upload.upload_id,
"status": upload.upload_status,
"upload_path": f"/internal/v1/uploads/{upload.upload_id}",
"expires_at": upload.expires_at.isoformat(),
}
# ── upload_bytes_to_session (server-proxied PUT) ────────────────────────
async def upload_bytes_to_session(
upload_id: str,
session: AsyncSession,
request: Request,
) -> StorageObjects:
"""Read raw bytes from the request body, validate against the
UploadSessions row, write via AsyncStorageBackend.put, and build the
StorageObjects row. Returns the row (caller may serialize it).
"""
upload = await session.scalar(
select(UploadSessions)
.where(UploadSessions.upload_id == upload_id)
.with_for_update()
)
if upload is None:
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 or item.object_status != "available":
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,
f"upload cannot continue from status {upload.upload_status}",
)
if upload.expires_at < _utcnow_naive():
upload.upload_status = "expired"
raise HTTPException(status.HTTP_409_CONFLICT, "upload expired")
content = await request.body()
actual_size = len(content)
if (
upload.expected_size_bytes is not None
and actual_size != upload.expected_size_bytes
):
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded bytes size does not match expected_size_bytes",
)
actual_hash = hashlib.sha256(content).hexdigest() if content else ""
if upload.expected_hash and actual_hash != upload.expected_hash:
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded bytes hash does not match expected_hash",
)
s3_metadata: dict[str, str] = {}
if actual_hash:
s3_metadata["sha256"] = actual_hash
try:
await request.app.state.object_stores[upload.bucket_name].put(
upload.object_key,
content,
content_type=upload.content_type,
metadata=s3_metadata or None,
)
except Exception as exc:
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
f"failed to write object to storage: {exc}",
) from exc
item = _build_storage_object(
upload=upload,
file_name=upload.file_name,
content_type=upload.content_type,
size_bytes=actual_size,
content_hash=actual_hash or None,
visibility=upload.visibility,
is_immutable=bool(upload.is_immutable),
usage_type=upload.usage_type,
)
session.add(item)
await session.flush()
await session.refresh(item)
upload.storage_object_id = item.storage_object_id
upload.upload_status = "completed"
upload.completed_at = _utcnow_naive()
return item
# ── create_server_object_payload ────────────────────────────────────────
async def create_server_object_payload(
payload: ServerObjectRequest,
request: Request,
session: AsyncSession,
) -> dict[str, Any]:
"""Server-side single-call upload (JSON body, base64 content).
Used by scripts.py when publishing version artifacts and by the
schedule worker for run logs / run results.
"""
from backend.storage_api import storage_payload
try:
content = base64.b64decode(payload.content_base64, validate=True)
except (binascii.Error, ValueError) as exc:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
"content_base64 is invalid",
) from exc
if len(content) > 100 * 1024 * 1024:
raise HTTPException(
status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
"object exceeds 100 MiB server-side upload limit",
)
content_hash = hashlib.sha256(content).hexdigest()
upload_result = await create_upload_record(
CreateUploadRequest(
workspace_id=payload.workspace_id,
user_id=payload.user_id,
usage_type=payload.usage_type,
file_name=payload.file_name,
content_type=payload.content_type,
expected_size_bytes=len(content),
expected_hash=content_hash,
idempotency_key=payload.idempotency_key,
visibility=payload.visibility,
is_immutable=payload.is_immutable,
),
session,
request,
)
if upload_result.get("status") == "completed":
existing_data = upload_result["storage_object"]
if (
payload.relative_path
and existing_data
and existing_data.get("relative_path") != payload.relative_path
):
existing_item = await session.get(
StorageObjects, existing_data["storage_object_id"]
)
if existing_item is not None:
existing_item.relative_path = payload.relative_path
await session.flush()
existing_data = storage_payload(existing_item)
return {"data": existing_data, "meta": {"reused": True}}
upload = await session.get(UploadSessions, upload_result["upload_id"])
if upload is None:
raise HTTPException(
status.HTTP_500_INTERNAL_SERVER_ERROR,
"upload record disappeared",
)
try:
await request.app.state.object_stores[upload.bucket_name].put(
upload.object_key,
content,
content_type=payload.content_type,
metadata={"sha256": content_hash},
)
except Exception as exc:
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
f"failed to write object to storage: {exc}",
) from exc
item = _build_storage_object(
upload=upload,
file_name=payload.file_name,
content_type=payload.content_type,
size_bytes=len(content),
content_hash=content_hash,
visibility=payload.visibility,
is_immutable=payload.is_immutable,
usage_type=payload.usage_type,
)
session.add(item)
await session.flush()
item.relative_path = payload.relative_path
upload.storage_object_id = item.storage_object_id
upload.upload_status = "completed"
upload.completed_at = _utcnow_naive()
return {"data": storage_payload(item), "meta": {"reused": False}}
# ── create_download_url_payload ─────────────────────────────────────────
async def create_download_url_payload(
item: StorageObjects,
payload: DownloadUrlRequest,
request: Request,
) -> dict[str, Any]:
"""Build a presigned GET URL for one StorageObjects row.
The caller (route handler in storage_api / scripts.py / resources.py)
loads the StorageObjects row + validates ownership/visibility; this
helper just builds the URL.
"""
if item is None or item.object_status != "available":
raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found")
if (
item.storage_backend != "s3"
or not item.bucket_name
or not item.object_key
):
raise HTTPException(
status.HTTP_409_CONFLICT,
"object does not support a presigned URL",
)
url = await request.app.state.object_stores[item.bucket_name].get_url(
item.object_key,
expires_in=timedelta(seconds=payload.expires_seconds),
)
# Public-host rewriting is now nginx's job (location /storage/). In the
# future the boto3 client should be built with the public endpoint so
# generate_presigned_url returns a public URL directly.
return {
"data": {
"storage_object_id": item.storage_object_id,
"presigned_url": url,
"method": "GET",
"expires_in_seconds": payload.expires_seconds,
}
}
# ── soft_delete_object ──────────────────────────────────────────────────
async def soft_delete_object(
storage_object_id: str,
request: Request,
session: AsyncSession,
) -> dict[str, Any]:
"""Soft-delete a storage object: copy to trash bucket, delete source,
flip the row to ``"deleted"``. Immutable objects are rejected.
"""
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.is_immutable:
raise HTTPException(
status.HTTP_409_CONFLICT,
"immutable object cannot be deleted",
)
if item.object_status == "deleted":
return {
"data": {
"storage_object_id": storage_object_id,
"object_status": item.object_status,
"trash_key": item.trash_key,
"trash_bucket": settings.s3_trash_bucket,
}
}
if item.storage_backend == "s3" and item.bucket_name and item.object_key:
trash_key = f"{item.bucket_name}/{item.object_key}"
try:
object_stores = request.app.state.object_stores
data = await object_stores[item.bucket_name].get(item.object_key)
await object_stores[settings.s3_trash_bucket].put(trash_key, data)
await object_stores[item.bucket_name].delete(item.object_key)
except Exception as exc:
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_naive()
return {
"data": {
"storage_object_id": storage_object_id,
"object_status": item.object_status,
"trash_key": item.trash_key,
"trash_bucket": settings.s3_trash_bucket,
}
}
backend/src/backend/storage_api.py
+184 -358
View File
@@ -1,14 +1,8 @@
from __future__ import annotations
import asyncio
import base64
import binascii
import hashlib
import mimetypes
import secrets
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from pathlib import PurePosixPath
from typing import Any, AsyncIterator
from fastapi import Depends, HTTPException, Request, status
@@ -22,15 +16,29 @@ from common.db.models import (
UploadSessions,
Users,
WorkspaceMembers,
Workspaces)
Workspaces,
)
from common.ids import new_ulid
from common.service_app import create_service_app
from common.storage import RustFSObjectStore
from common.storage import (
AsyncStorageBackend,
PURPOSE_BUCKETS,
actual_bucket_name,
build_storage_config,
create_storage,
)
from common.storage.schemas import (
CompleteUploadRequest,
CreateUploadRequest,
DownloadUrlRequest,
ServerObjectRequest)
ServerObjectRequest,
)
from backend.services.storage import (
create_download_url_payload,
create_server_object_payload,
create_upload_record,
soft_delete_object,
upload_bytes_to_session,
)
def utcnow() -> datetime:
@@ -41,10 +49,7 @@ def hash_bytes(value: str) -> bytes:
return hashlib.sha256(value.encode("utf-8")).digest()
def normalized_idempotency_key(
workspace_id: str,
user_id: str,
value: str) -> str:
def normalized_idempotency_key(workspace_id: str, user_id: str, value: str) -> str:
digest = hashlib.sha256(
f"{workspace_id}:{user_id}:{value}".encode("utf-8")
).hexdigest()
@@ -58,20 +63,26 @@ def safe_file_name(value: str) -> str:
return name
# Map an upload's usage_type to the RustFS bucket that should hold the
# resulting object. ``usage_type`` is the only signal available at the
# storage edge (the request comes from either the public API or the
# internal schedule worker), so we make the routing decision in one place
# here and let every other layer — server-object create, multipart upload,
# direct put — inherit the mapping.
# Map an upload's usage_type to its purpose (which then resolves to the
# actual bucket / directory via ``actual_bucket_name``). Keeping the
# purpose as the intermediate value means s3 mode and local mode share
# the same routing logic — only the final ``actual_bucket_name`` differs.
USAGE_TYPE_TO_PURPOSE: dict[str, str] = {
"working_copy": "workspace",
"public_script": "workspace",
"data_resource": "workspace",
"snapshot": "workspace",
"version_artifact": "version",
"run_log": "run_log",
"run_result": "run_log",
}
# Pre-resolved bucket map (for read-only callers like services/storage.py).
# Re-resolved at module load; re-resolve via resolve_bucket() if the
# workspace.artifact_bucket override matters.
BUCKET_FOR_USAGE: dict[str, str] = {
"working_copy": settings.rustfs_workspace_bucket,
"public_script": settings.rustfs_workspace_bucket,
"data_resource": settings.rustfs_workspace_bucket,
"snapshot": settings.rustfs_workspace_bucket,
"version_artifact": settings.rustfs_version_bucket,
"run_log": settings.rustfs_run_log_bucket,
"run_result": settings.rustfs_run_log_bucket,
usage_type: actual_bucket_name(purpose)
for usage_type, purpose in USAGE_TYPE_TO_PURPOSE.items()
}
@@ -88,7 +99,8 @@ def resolve_bucket(
"""
if workspace.artifact_bucket:
return workspace.artifact_bucket
return BUCKET_FOR_USAGE.get(usage_type, settings.rustfs_workspace_bucket)
purpose = USAGE_TYPE_TO_PURPOSE.get(usage_type, "workspace")
return actual_bucket_name(purpose)
def storage_payload(item: StorageObjects) -> dict[str, Any]:
@@ -118,35 +130,23 @@ def storage_payload(item: StorageObjects) -> dict[str, Any]:
async def lifespan(app: Any) -> AsyncIterator[None]:
engine = create_database_engine(settings.database_url)
app.state.session_factory = create_session_factory(engine)
app.state.object_store = RustFSObjectStore(
internal_endpoint=settings.rustfs_endpoint,
access_key=settings.rustfs_access_key,
secret_key=settings.rustfs_secret_key,
)
app.state.default_bucket = settings.rustfs_workspace_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,
)
# Buckets are pre-provisioned by the deployment; the storage layer no
# longer auto-creates them. ``build_storage_config`` picks s3 vs local
# based on ``settings.storage_backend``. The dict is keyed by the
# actual bucket name so ``object_stores[upload.bucket_name]``
# works without a reverse mapping.
app.state.object_stores: dict[str, AsyncStorageBackend] = {
actual_bucket_name(purpose): create_storage(build_storage_config(purpose))
for purpose in PURPOSE_BUCKETS
}
app.state.default_bucket = settings.s3_workspace_bucket
try:
yield
finally:
await engine.dispose()
app = create_service_app(
settings.service_name,
lifespan=lifespan)
app = create_service_app(settings.service_name, lifespan=lifespan)
async def database_session(request: Request) -> AsyncIterator[AsyncSession]:
@@ -155,46 +155,41 @@ async def database_session(request: Request) -> AsyncIterator[AsyncSession]:
async def require_workspace_member(
session: AsyncSession,
workspace_id: str,
user_id: str) -> Workspaces:
session: AsyncSession, workspace_id: str, user_id: str
) -> Workspaces:
statement = (
select(Workspaces)
.join(
WorkspaceMembers,
WorkspaceMembers.workspace_id == Workspaces.workspace_id)
WorkspaceMembers, WorkspaceMembers.workspace_id == Workspaces.workspace_id
)
.join(Users, Users.user_id == WorkspaceMembers.user_id)
.where(
Workspaces.workspace_id == workspace_id,
Workspaces.status == "active",
WorkspaceMembers.user_id == user_id,
WorkspaceMembers.member_status == "active",
Users.status == "active")
Users.status == "active",
)
)
workspace = await session.scalar(statement)
if workspace is None:
raise HTTPException(
status.HTTP_403_FORBIDDEN,
"user is not an active workspace member")
status.HTTP_403_FORBIDDEN, "user is not an active workspace member"
)
return workspace
async def create_upload_record(
payload: CreateUploadRequest,
session: AsyncSession,
request: Request) -> dict[str, Any]:
payload: CreateUploadRequest, session: AsyncSession, request: Request
) -> dict[str, Any]:
workspace = await require_workspace_member(
session,
payload.workspace_id,
payload.user_id)
session, payload.workspace_id, payload.user_id
)
stored_key = normalized_idempotency_key(
payload.workspace_id,
payload.user_id,
payload.idempotency_key)
payload.workspace_id, payload.user_id, payload.idempotency_key
)
existing = await session.scalar(
select(UploadSessions).where(
UploadSessions.idempotency_key == stored_key
)
select(UploadSessions).where(UploadSessions.idempotency_key == stored_key)
)
if existing is not None:
if (
@@ -206,7 +201,8 @@ async def create_upload_record(
):
raise HTTPException(
status.HTTP_409_CONFLICT,
"idempotency key was used with different upload metadata")
"idempotency key was used with different upload metadata",
)
upload = existing
else:
upload_id = new_ulid()
@@ -214,9 +210,7 @@ async def create_upload_record(
# Keep the opaque upload id while preserving the original extension.
# Jupyter selects its editor from this suffix, so an extensionless
# object would make notebooks look like generic JSON/text files.
file_extension = PurePosixPath(
safe_file_name(payload.file_name)
).suffix.lower()
file_extension = PurePosixPath(safe_file_name(payload.file_name)).suffix.lower()
object_key = f"{payload.workspace_id}/{upload_id}{file_extension}"
upload = UploadSessions(
upload_id=upload_id,
@@ -230,14 +224,17 @@ async def create_upload_record(
expires_at=utcnow() + timedelta(minutes=15),
expected_size_bytes=payload.expected_size_bytes,
expected_hash=payload.expected_hash,
content_type=payload.content_type)
content_type=payload.content_type,
file_name=payload.file_name,
usage_type=payload.usage_type,
visibility=payload.visibility,
is_immutable=int(payload.is_immutable),
)
session.add(upload)
await session.flush()
if upload.upload_status == "completed" and upload.storage_object_id:
storage_object = await session.get(
StorageObjects,
upload.storage_object_id)
storage_object = await session.get(StorageObjects, upload.storage_object_id)
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
@@ -254,24 +251,16 @@ async def create_upload_record(
if upload.upload_status not in {"created", "uploading"}:
raise HTTPException(
status.HTTP_409_CONFLICT,
f"upload cannot continue from status {upload.upload_status}")
f"upload cannot continue from status {upload.upload_status}",
)
url, headers = request.app.state.object_store.presign_put(
bucket_name=upload.bucket_name,
object_key=upload.object_key,
content_type=upload.content_type or "application/octet-stream",
expected_hash=upload.expected_hash,
expires_seconds=900)
presigned_url = request.app.state.object_store.rewrite_to_public_path(
url,
public_base_url=_public_base_url(request),
)
# Two-step server-proxied upload: the caller PUTs the raw bytes to
# ``upload_path`` after this response, which routes through
# ``upload_bytes_to_session`` below.
return {
"upload_id": upload.upload_id,
"status": upload.upload_status,
"method": "PUT",
"presigned_url": presigned_url,
"required_headers": headers,
"upload_path": f"/internal/v1/uploads/{upload.upload_id}",
"expires_at": upload.expires_at.isoformat(),
}
@@ -282,7 +271,7 @@ def _public_base_url(request: Request) -> str:
Falls back to the inbound request's ``Host`` header and the scheme
Nginx forwards via ``X-Forwarded-Proto`` so the resulting
presigned URL always points at the public edge rather than the
in-cluster RustFS endpoint.
in-cluster S3 endpoint.
"""
forwarded_proto = request.headers.get("x-forwarded-proto", "").strip()
scheme = forwarded_proto or request.url.scheme or "http"
@@ -298,11 +287,15 @@ def _public_base_url(request: Request) -> str:
return f"{scheme}://{host}"
async def complete_upload_record(
upload_id: str,
payload: CompleteUploadRequest,
session: AsyncSession,
request: Request) -> StorageObjects:
async def upload_bytes_to_session(
upload_id: str, session: AsyncSession, request: Request
) -> StorageObjects:
"""Server-proxied upload: read raw bytes from the request body, validate
against the ``UploadSessions`` expectations, call ``backend.put``, and
create the ``StorageObjects`` row.
Replaces the old presign-PUT + head-validate flow.
"""
upload = await session.scalar(
select(UploadSessions)
.where(UploadSessions.upload_id == upload_id)
@@ -313,9 +306,7 @@ async def complete_upload_record(
if upload.upload_status == "completed" and upload.storage_object_id:
item = await session.get(StorageObjects, upload.storage_object_id)
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.
# Linked object was deleted; allow re-upload with the same id.
upload.storage_object_id = None
upload.upload_status = "created"
else:
@@ -323,22 +314,15 @@ async def complete_upload_record(
if upload.upload_status not in {"created", "uploading"}:
raise HTTPException(
status.HTTP_409_CONFLICT,
f"upload cannot be completed from status {upload.upload_status}")
f"upload cannot continue from status {upload.upload_status}",
)
if upload.expires_at < utcnow():
upload.upload_status = "expired"
raise HTTPException(status.HTTP_409_CONFLICT, "upload expired")
try:
head = await asyncio.to_thread(
request.app.state.object_store.head,
bucket_name=upload.bucket_name,
object_key=upload.object_key)
except Exception as exc:
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded object is not available") from exc
content = await request.body()
actual_size = len(content)
actual_size = int(head.get("ContentLength", 0))
if (
upload.expected_size_bytes is not None
and actual_size != upload.expected_size_bytes
@@ -346,57 +330,59 @@ async def complete_upload_record(
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded object size does not match expected_size_bytes")
actual_content_type = str(
head.get("ContentType") or "application/octet-stream"
)
if upload.content_type and actual_content_type != upload.content_type:
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded object content type does not match")
metadata = {
str(key).lower(): str(value).lower()
for key, value in dict(head.get("Metadata") or {}).items()
}
actual_hash = metadata.get("sha256")
if not actual_hash:
actual_hash = await asyncio.to_thread(
request.app.state.object_store.sha256,
bucket_name=upload.bucket_name,
object_key=upload.object_key)
"uploaded bytes size does not match expected_size_bytes",
)
actual_hash = hashlib.sha256(content).hexdigest() if content else ""
if upload.expected_hash and actual_hash != upload.expected_hash:
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded object hash does not match expected_hash")
status.HTTP_409_CONFLICT, "uploaded bytes hash does not match expected_hash"
)
# The object key is just ``{workspace_id}/{ulid}`` — it does not encode
# the file name. Use the original file name from the upload session
# (carried via payload.file_name) so the StorageObjects row still
# records the user-visible name + extension.
file_name = safe_file_name(payload.file_name)
# Round-trip content_type + sha256 metadata through the storage backend
# so the next head() (or our own put signature) can recover them.
s3_metadata: dict[str, str] = {}
if actual_hash:
s3_metadata["sha256"] = actual_hash
try:
await request.app.state.object_stores[upload.bucket_name].put(
upload.object_key,
content,
content_type=upload.content_type,
metadata=s3_metadata or None,
)
except Exception as exc:
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
f"failed to write object to storage: {exc}",
) from exc
file_name = safe_file_name(upload.file_name_hint or "upload.bin")
item = StorageObjects(
storage_object_id=new_ulid(),
workspace_id=upload.workspace_id,
owner_user_id=upload.user_id,
object_type="file",
usage_type=payload.usage_type,
storage_backend="rustfs",
usage_type=upload.usage_type,
storage_backend="s3",
bucket_name=upload.bucket_name,
object_key=upload.object_key,
object_key_hash=upload.object_key_hash,
storage_uri=f"s3://{upload.bucket_name}/{upload.object_key}",
file_name=file_name,
file_extension=PurePosixPath(file_name).suffix.lower() or None,
mime_type=actual_content_type,
mime_type=upload.content_type,
size_bytes=actual_size,
content_hash=actual_hash,
object_etag=str(head.get("ETag", "")).strip('"') or None,
visibility=payload.visibility,
is_immutable=int(payload.is_immutable),
content_hash=actual_hash or None,
object_etag=None,
visibility=upload.visibility,
is_immutable=int(upload.is_immutable),
object_status="available",
created_by=upload.user_id)
created_by=upload.user_id,
)
session.add(item)
await session.flush()
await session.refresh(item)
@@ -410,33 +396,29 @@ async def complete_upload_record(
async def create_upload(
payload: CreateUploadRequest,
request: Request,
session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
return {
"data": await create_upload_record(payload, session, request),
}
@app.post(
"/internal/v1/uploads/{upload_id}/complete")
async def complete_upload(
upload_id: str,
payload: CompleteUploadRequest,
request: Request,
session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
item = await complete_upload_record(
upload_id,
payload,
session,
request)
@app.put("/internal/v1/uploads/{upload_id}")
async def upload_bytes(
upload_id: str, request: Request, session: AsyncSession = Depends(database_session)
) -> dict[str, Any]:
"""Server-proxied upload: PUT raw bytes in the request body. Replaces the
old ``POST /uploads/{id}/complete`` flow that paired presigned-PUT with
a head()-validate step.
"""
item = await upload_bytes_to_session(upload_id, session, request)
return {"data": storage_payload(item)}
@app.post(
"/internal/v1/uploads/{upload_id}/abort")
@app.post("/internal/v1/uploads/{upload_id}/abort")
async def abort_upload(
upload_id: str,
request: Request,
session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
upload_id: str, request: Request, session: AsyncSession = Depends(database_session)
) -> dict[str, Any]:
upload = await session.scalar(
select(UploadSessions)
.where(UploadSessions.upload_id == upload_id)
@@ -446,202 +428,52 @@ async def abort_upload(
raise HTTPException(status.HTTP_404_NOT_FOUND, "upload not found")
if upload.upload_status == "completed":
raise HTTPException(
status.HTTP_409_CONFLICT,
"completed upload cannot be aborted")
status.HTTP_409_CONFLICT, "completed upload cannot be aborted"
)
if upload.upload_status != "aborted":
await asyncio.to_thread(
request.app.state.object_store.delete,
bucket_name=upload.bucket_name,
object_key=upload.object_key)
await request.app.state.object_stores[upload.bucket_name].delete(
upload.object_key
)
upload.upload_status = "aborted"
return {"data": {"upload_id": upload_id, "status": "aborted"}}
@app.post(
"/internal/v1/objects")
@app.post("/internal/v1/objects")
async def create_server_object(
payload: ServerObjectRequest,
request: Request,
session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
try:
content = base64.b64decode(payload.content_base64, validate=True)
except (binascii.Error, ValueError) as exc:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
"content_base64 is invalid") from exc
if len(content) > 100 * 1024 * 1024:
raise HTTPException(
status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
"object exceeds 100 MiB server-side upload limit")
content_hash = hashlib.sha256(content).hexdigest()
upload_result = await create_upload_record(
CreateUploadRequest(
workspace_id=payload.workspace_id,
user_id=payload.user_id,
usage_type=payload.usage_type,
file_name=payload.file_name,
content_type=payload.content_type,
expected_size_bytes=len(content),
expected_hash=content_hash,
idempotency_key=payload.idempotency_key),
session,
request)
if upload_result.get("status") == "completed":
existing_data = upload_result["storage_object"]
if (
payload.relative_path
and existing_data
and existing_data.get("relative_path") != payload.relative_path
):
existing_item = await session.get(
StorageObjects,
existing_data["storage_object_id"],
)
if existing_item is not None:
existing_item.relative_path = payload.relative_path
await session.flush()
existing_data = storage_payload(existing_item)
return {"data": existing_data, "meta": {"reused": True}}
upload = await session.get(UploadSessions, upload_result["upload_id"])
if upload is None:
raise HTTPException(
status.HTTP_500_INTERNAL_SERVER_ERROR,
"upload record disappeared")
await asyncio.to_thread(
request.app.state.object_store.put_bytes,
bucket_name=upload.bucket_name,
object_key=upload.object_key,
content=content,
content_type=payload.content_type,
content_hash=content_hash)
item = await complete_upload_record(
upload.upload_id,
CompleteUploadRequest(
usage_type=payload.usage_type,
file_name=payload.file_name,
visibility=payload.visibility,
is_immutable=payload.is_immutable),
session,
request)
item.relative_path = payload.relative_path
await session.flush()
return {"data": storage_payload(item), "meta": {"reused": False}}
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
return await create_server_object_payload(payload, request, session)
@app.post(
"/internal/v1/objects/{storage_object_id}/download-url")
@app.post("/internal/v1/objects/{storage_object_id}/download-url")
async def create_download_url(
storage_object_id: str,
payload: DownloadUrlRequest,
request: Request,
session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
item = await session.get(StorageObjects, storage_object_id)
if item is None or item.object_status != "available":
raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found")
if (
item.storage_backend != "rustfs"
or not item.bucket_name
or not item.object_key
):
raise HTTPException(
status.HTTP_409_CONFLICT,
"object does not support a presigned URL")
url = request.app.state.object_store.presign_get(
bucket_name=item.bucket_name,
object_key=item.object_key,
file_name=item.file_name,
expires_seconds=payload.expires_seconds)
presigned_url = request.app.state.object_store.rewrite_to_public_path(
url,
public_base_url=_public_base_url(request),
)
return {
"data": {
"storage_object_id": item.storage_object_id,
"presigned_url": presigned_url,
"method": "GET",
"expires_in_seconds": payload.expires_seconds,
}
}
return await create_download_url_payload(item, payload, request)
@app.delete(
"/internal/v1/objects/{storage_object_id}")
@app.delete("/internal/v1/objects/{storage_object_id}")
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)
.with_for_update()
)
if item is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found")
if item.is_immutable:
raise HTTPException(
status.HTTP_409_CONFLICT,
"immutable object cannot be deleted")
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.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,
}
}
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Soft-delete a storage object. See ``backend.services.storage.soft_delete_object``."""
return await soft_delete_object(storage_object_id, request, session)
@app.post(
"/internal/v1/objects/{storage_object_id}/restore")
@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]:
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
@@ -660,21 +492,16 @@ async def restore_object(
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")
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,
status.HTTP_409_CONFLICT, "object has no trash pointer; cannot restore"
)
try:
# Cross-backend copy: get from trash, put back to source bucket.
object_stores = request.app.state.object_stores
data = await object_stores[settings.s3_trash_bucket].get(item.trash_key)
await object_stores[item.bucket_name].put(item.object_key, data)
except Exception as exc:
raise HTTPException(
status.HTTP_502_BAD_GATEWAY,
@@ -690,12 +517,12 @@ async def restore_object(
}
@app.post(
"/internal/v1/admin/trash/purge")
@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]:
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""Physically delete a trashed object.
Admin / reaper endpoint — given a ``storage_object_id``, deletes
@@ -707,8 +534,8 @@ async def purge_trash_object(
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")
status.HTTP_400_BAD_REQUEST, "storage_object_id is required"
)
item = await session.scalar(
select(StorageObjects)
.where(StorageObjects.storage_object_id == storage_object_id)
@@ -719,13 +546,12 @@ async def purge_trash_object(
if item.object_status != "deleted":
raise HTTPException(
status.HTTP_409_CONFLICT,
"object is not in trash; refuse to hard-delete live data")
"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,
await request.app.state.object_stores[settings.s3_trash_bucket].delete(
item.trash_key
)
except Exception as exc:
raise HTTPException(
backend/src/backend/storage_client.py
-66
View File
@@ -1,66 +0,0 @@
"""Backend-bound storage client.
Re-exports :class:`StorageClient` under the same name used by callers in
``backend/``. The default client raises :class:`StorageClientError` from
``common.storage.client`` so it stays usable from non-FastAPI contexts.
Inside FastAPI route handlers we want HTTP-shaped errors, so this module
also exposes :class:`BackendStorageClient`, a thin wrapper that translates
the framework-agnostic errors into ``HTTPException``.
"""
from __future__ import annotations
from typing import Any
from fastapi import HTTPException, status
from common.storage.client import (
StorageClient,
StorageClientError,
StorageRequestFailed,
StorageUnavailable,
)
__all__ = ["BackendStorageClient", "StorageClient", "StorageClientError"]
def _to_http_exception(exc: StorageClientError) -> HTTPException:
if isinstance(exc, StorageUnavailable):
return HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
{
"code": "STORAGE_UNAVAILABLE",
"message": "Storage service temporarily unavailable",
"retryable": True,
"details": {},
},
)
if isinstance(exc, StorageRequestFailed):
return HTTPException(exc.status_code, exc.detail)
return HTTPException(
status.HTTP_500_INTERNAL_SERVER_ERROR,
"Storage client error",
)
class BackendStorageClient(StorageClient):
"""Storage client that raises ``HTTPException`` for web callers."""
async def _request(
self,
method: str,
path: str,
*,
payload: dict[str, Any] | None = None,
) -> dict[str, Any]:
try:
return await super()._request(method, path, payload=payload)
except StorageClientError as exc:
raise _to_http_exception(exc) from exc
# Re-bind the imported symbol so existing backend call sites that import
# ``StorageClient`` from this module transparently get the FastAPI-bound
# variant without changing every import statement.
StorageClient = BackendStorageClient