From f4cabe6e205f74ae45315a9a2c52b9da6cd7fbe9 Mon Sep 17 00:00:00 2001 From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com> Date: Wed, 5 Aug 2026 13:41:13 +0800 Subject: [PATCH] update: remove storage --- backend/src/backend/main.py | 17 +- backend/src/backend/resources.py | 132 ++++-- backend/src/backend/scripts.py | 44 +- backend/src/backend/services/__init__.py | 0 backend/src/backend/services/storage.py | 512 +++++++++++++++++++++++ backend/src/backend/storage_api.py | 192 +-------- backend/src/backend/storage_client.py | 72 ---- schedule/src/schedule/storage_client.py | 46 +- 8 files changed, 683 insertions(+), 332 deletions(-) create mode 100644 backend/src/backend/services/__init__.py create mode 100644 backend/src/backend/services/storage.py delete mode 100644 backend/src/backend/storage_client.py diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 54d9d1c..4b77137 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -22,7 +22,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 @@ -30,21 +29,14 @@ 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. - # 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`` — see common/storage/factory.py. + # 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. ``build_storage_config`` picks s3 vs + # local based on ``settings.storage_backend``. app.state.object_stores: dict[str, AsyncStorageBackend] = { name: create_storage(build_storage_config(name)) for name in PURPOSE_BUCKETS } app.state.default_bucket = settings.s3_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) runtime_http_client = httpx.AsyncClient( base_url=settings.runtime_api_url, timeout=httpx.Timeout(30.0), @@ -61,7 +53,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() diff --git a/backend/src/backend/resources.py b/backend/src/backend/resources.py index 0c9dab1..e57b61e 100644 --- a/backend/src/backend/resources.py +++ b/backend/src/backend/resources.py @@ -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 { diff --git a/backend/src/backend/scripts.py b/backend/src/backend/scripts.py index 267e7e4..dbe8940 100644 --- a/backend/src/backend/scripts.py +++ b/backend/src/backend/scripts.py @@ -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, @@ -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": {}} diff --git a/backend/src/backend/services/__init__.py b/backend/src/backend/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/src/backend/services/storage.py b/backend/src/backend/services/storage.py new file mode 100644 index 0000000..0789b43 --- /dev/null +++ b/backend/src/backend/services/storage.py @@ -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, + } + } \ No newline at end of file diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index 4dff846..d3b618d 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -1,14 +1,10 @@ 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 @@ -30,6 +26,13 @@ from common.storage.schemas import ( CreateUploadRequest, DownloadUrlRequest, 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: @@ -435,97 +438,7 @@ 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") - - # Write directly via the storage backend; metadata + content_type are - # carried through so a subsequent head() can recover them. The - # StorageObjects row construction mirrors upload_bytes_to_session. - 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 - - file_name = safe_file_name(payload.file_name) - 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="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=payload.content_type, - size_bytes=len(content), - content_hash=content_hash, - object_etag=None, - visibility=payload.visibility, - is_immutable=int(payload.is_immutable), - object_status="available", - created_by=upload.user_id) - 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() - return {"data": storage_payload(item), "meta": {"reused": False}} + return await create_server_object_payload(payload, request, session) @app.post( @@ -536,32 +449,7 @@ async def create_download_url( request: Request, 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 != "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. - presigned_url = url - 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( @@ -570,66 +458,8 @@ 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 ``s3_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 ``s3_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 == "s3" and item.bucket_name and item.object_key: - trash_key = f"{item.bucket_name}/{item.object_key}" - try: - # Cross-backend move: get from source, put to trash, delete source. - 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: - # 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.s3_trash_bucket, - } - } + """Soft-delete a storage object. See ``backend.services.storage.soft_delete_object``.""" + return await soft_delete_object(storage_object_id, request, session) @app.post( diff --git a/backend/src/backend/storage_client.py b/backend/src/backend/storage_client.py deleted file mode 100644 index 0cb86d1..0000000 --- a/backend/src/backend/storage_client.py +++ /dev/null @@ -1,72 +0,0 @@ -"""Backend-bound storage client. - -TODO: this HTTP client is dead code post-migration; rewrite to use -AsyncStorageBackend directly. The base StorageClient class was removed -from common.storage.client, so this module is currently a stub that -preserves the import surface but raises NotImplementedError. -""" - -from __future__ import annotations - -from typing import Any - -from fastapi import HTTPException, status - - -__all__ = ["BackendStorageClient", "StorageClient", "StorageClientError"] - - -class StorageClientError(Exception): - pass - - -class StorageUnavailable(StorageClientError): - pass - - -class StorageRequestFailed(StorageClientError): - def __init__(self, status_code: int, detail: Any): - self.status_code = status_code - self.detail = detail - super().__init__(status_code, detail) - - -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: - """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]: - raise NotImplementedError( - "TODO: BackendStorageClient is dead code post-migration; " - "rewrite to use AsyncStorageBackend directly" - ) - - -# 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 diff --git a/schedule/src/schedule/storage_client.py b/schedule/src/schedule/storage_client.py index b4b74a5..0abc7ff 100644 --- a/schedule/src/schedule/storage_client.py +++ b/schedule/src/schedule/storage_client.py @@ -1,15 +1,27 @@ -"""Schedule-specific storage client built on the shared ``StorageClient``.""" +"""Schedule-side HTTP client for the backend's storage API. + +The schedule worker uploads run logs / run results by calling +``POST {backend}/internal/v1/objects`` (the backend's +``create_server_object_payload`` route, which is in the same process +as the public API). The response is the StorageObjects row payload. + +The schedule does NOT have its own DB session for storage metadata, +so it must go through the backend to create the StorageObjects row +(``logs_object_id`` / ``result_object_id`` are FKs into that table). +""" from __future__ import annotations +import base64 from typing import Any -# TODO: common.storage.client.StorageClient was removed in the S3 migration. -# This module is temporary dead code; rewrite to use AsyncStorageBackend. -# from common.storage.client import StorageClient +import httpx class SchedulerStorageClient: + def __init__(self, http_client: httpx.AsyncClient) -> None: + self._http = http_client + async def create_object( self, *, @@ -21,10 +33,30 @@ class SchedulerStorageClient: content: bytes, idempotency_key: str, ) -> dict[str, Any]: - raise NotImplementedError( - "TODO: SchedulerStorageClient is dead code post-migration; " - "rewrite to use AsyncStorageBackend directly" + """Upload a run_log / run_result via the backend's storage API. + + The backend returns ``{"data": , "meta": {...}}``; + we return the inner ``data`` dict (which includes + ``storage_object_id`` and ``storage_uri``). + """ + response = await self._http.post( + "/internal/v1/objects", + json={ + "workspace_id": workspace_id, + "user_id": user_id, + "usage_type": usage_type, + "file_name": file_name, + "content_type": content_type, + "content_base64": base64.b64encode(content).decode("ascii"), + "visibility": "workspace", + "is_immutable": True, + "idempotency_key": idempotency_key, + "relative_path": None, + }, ) + response.raise_for_status() + body = response.json() + return body["data"] __all__ = ["SchedulerStorageClient"]