update: remove workspace operation table and refactor

backend/src/backend/storage_api.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
-import asyncio
-import mimetypes
-import secrets
+import hashlib
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime, timedelta
 from typing import Any, AsyncIterator
@@ -18,14 +16,21 @@ 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 AsyncStorageBackend, PURPOSE_BUCKETS, build_storage_config, create_storage
+from common.storage import (
+    AsyncStorageBackend,
+    PURPOSE_BUCKETS,
+    build_storage_config,
+    create_storage,
+)
 from common.storage.schemas import (
     CreateUploadRequest,
     DownloadUrlRequest,
-    ServerObjectRequest)
+    ServerObjectRequest,
+)
 from backend.services.storage import (
     create_download_url_payload,
     create_server_object_payload,
@@ -43,10 +48,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()
@@ -133,9 +135,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
         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]:
@@ -144,46 +144,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 (
@@ -195,7 +190,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()
@@ -203,9 +199,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,
@@ -223,14 +217,13 @@ async def create_upload_record(
             file_name=payload.file_name,
             usage_type=payload.usage_type,
             visibility=payload.visibility,
-            is_immutable=int(payload.is_immutable))
+            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
@@ -247,7 +240,8 @@ 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}",
+        )
 
     # Two-step server-proxied upload: the caller PUTs the raw bytes to
     # ``upload_path`` after this response, which routes through
@@ -283,9 +277,8 @@ def _public_base_url(request: Request) -> str:
 
 
 async def upload_bytes_to_session(
-    upload_id: str,
-    session: AsyncSession,
-    request: Request) -> StorageObjects:
+    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.
@@ -310,7 +303,8 @@ async def upload_bytes_to_session(
     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}",
+        )
     if upload.expires_at < utcnow():
         upload.upload_status = "expired"
         raise HTTPException(status.HTTP_409_CONFLICT, "upload expired")
@@ -325,14 +319,15 @@ async def upload_bytes_to_session(
         upload.upload_status = "failed"
         raise HTTPException(
             status.HTTP_409_CONFLICT,
-            "uploaded bytes size does not match expected_size_bytes")
+            "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")
+            status.HTTP_409_CONFLICT, "uploaded bytes hash does not match expected_hash"
+        )
 
     # Round-trip content_type + sha256 metadata through the storage backend
     # so the next head() (or our own put signature) can recover them.
@@ -351,7 +346,8 @@ async def upload_bytes_to_session(
         upload.upload_status = "failed"
         raise HTTPException(
             status.HTTP_503_SERVICE_UNAVAILABLE,
-            f"failed to write object to storage: {exc}") from exc
+            f"failed to write object to storage: {exc}",
+        ) from exc
 
     file_name = safe_file_name(upload.file_name_hint or "upload.bin")
     item = StorageObjects(
@@ -374,7 +370,8 @@ async def upload_bytes_to_session(
         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)
@@ -388,7 +385,8 @@ async def upload_bytes_to_session(
 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),
     }
@@ -396,9 +394,8 @@ async def create_upload(
 
 @app.put("/internal/v1/uploads/{upload_id}")
 async def upload_bytes(
-    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]:
     """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.
@@ -407,12 +404,10 @@ async def upload_bytes(
     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)
@@ -422,52 +417,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 request.app.state.object_stores[
-            upload.bucket_name
-        ].delete(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]:
+    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)
     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]:
+    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
@@ -486,13 +481,11 @@ 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")
+            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
@@ -513,12 +506,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
@@ -530,8 +523,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)
@@ -542,12 +535,13 @@ 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 request.app.state.object_stores[
-                settings.s3_trash_bucket
-            ].delete(item.trash_key)
+            await request.app.state.object_stores[settings.s3_trash_bucket].delete(
+                item.trash_key
+            )
         except Exception as exc:
             raise HTTPException(
                 status.HTTP_502_BAD_GATEWAY,