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

refactor

Browse Source
This commit is contained in:
tao.chen
2026-07-30 20:02:19 +08:00
parent c1e15758a3
commit ec53edbce5
23 changed files with 191 additions and 1085 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/backend/dependencies.py
+4 -8
View File
@@ -3,10 +3,11 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import AsyncIterator
from fastapi import Header, HTTPException, Request, status
from fastapi import Depends, Header, HTTPException, Request, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from common.db import session_scope
from common.db.models import (
Roles,
Users,
@@ -29,13 +30,8 @@ class RequestContext:
async def database_session(request: Request) -> AsyncIterator[AsyncSession]:
async with request.app.state.session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
async with session_scope(request.app.state.session_factory) as session:
yield session
async def request_context(
backend/src/backend/file_locks.py
-110
View File
@@ -1,110 +0,0 @@
from __future__ import annotations
from typing import Any, Awaitable, Callable
from fastapi import APIRouter, Depends, Request, status
from fastapi.responses import JSONResponse
from backend.dependencies import (
RequestContext,
request_context,
)
from backend.runtime_client import RuntimeClientError
from backend.schemas import FileLockTokenRequest
router = APIRouter(tags=["file-locks"])
async def runtime_response(
context: RequestContext,
operation: Callable[[], Awaitable[dict[str, Any]]],
*,
success_status: int = status.HTTP_200_OK,
) -> JSONResponse:
try:
data = await operation()
except RuntimeClientError as exc:
error = exc.detail
if not isinstance(error, dict) or "code" not in error:
error = {
"code": "RUNTIME_REQUEST_FAILED",
"message": str(error),
"retryable": exc.status_code >= 500,
"details": {},
}
return JSONResponse(
status_code=exc.status_code,
content={"request_id": context.request_id, "error": error},
)
return JSONResponse(
status_code=success_status,
content={
"request_id": context.request_id,
"data": data,
"meta": {},
},
)
@router.post(
"/api/v1/files/{storage_object_id}/lock",
status_code=status.HTTP_201_CREATED,
)
async def acquire_file_lock(
storage_object_id: str,
request: Request,
context: RequestContext = Depends(request_context),
) -> JSONResponse:
return await runtime_response(
context,
lambda: request.app.state.runtime_client.acquire_file_lock(
{
"workspace_id": context.workspace.workspace_id,
"storage_object_id": storage_object_id,
"user_id": context.user.user_id,
"request_id": context.request_id,
}
),
success_status=status.HTTP_201_CREATED,
)
@router.post("/api/v1/file-locks/{edit_session_id}/heartbeat")
async def heartbeat_file_lock(
edit_session_id: str,
payload: FileLockTokenRequest,
request: Request,
context: RequestContext = Depends(request_context),
) -> JSONResponse:
return await runtime_response(
context,
lambda: request.app.state.runtime_client.heartbeat_file_lock(
edit_session_id,
{
"workspace_id": context.workspace.workspace_id,
"user_id": context.user.user_id,
"lock_token": payload.lock_token,
},
),
)
@router.delete("/api/v1/file-locks/{edit_session_id}")
async def release_file_lock(
edit_session_id: str,
payload: FileLockTokenRequest,
request: Request,
context: RequestContext = Depends(request_context),
) -> JSONResponse:
return await runtime_response(
context,
lambda: request.app.state.runtime_client.release_file_lock(
edit_session_id,
{
"workspace_id": context.workspace.workspace_id,
"user_id": context.user.user_id,
"lock_token": payload.lock_token,
},
),
)
backend/src/backend/main.py
+2 -10
View File
@@ -13,7 +13,6 @@ from common.db import create_database_engine, create_session_factory
from common.service_app import create_service_app
from common.storage import RustFSObjectStore
from backend.admin import router as admin_router
from backend.file_locks import router as file_locks_router
from backend.jupyter import router as jupyter_router
from backend.resources import router as resources_router
from backend.runtime_client import RuntimeClient
@@ -60,18 +59,12 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
base_url="http://backend.internal",
timeout=httpx.Timeout(30.0),
)
app.state.storage_client = StorageClient(
storage_http_client,
os.environ["INTERNAL_SERVICE_TOKEN"],
)
app.state.storage_client = StorageClient(storage_http_client)
runtime_http_client = httpx.AsyncClient(
base_url=os.getenv("RUNTIME_API_URL", "http://runtime:8000"),
timeout=httpx.Timeout(30.0),
)
app.state.runtime_client = RuntimeClient(
runtime_http_client,
os.environ["INTERNAL_SERVICE_TOKEN"],
)
app.state.runtime_client = RuntimeClient(runtime_http_client)
try:
yield
finally:
@@ -84,7 +77,6 @@ app = create_service_app(
os.getenv("SERVICE_NAME", "backend"),
lifespan=lifespan,
)
app.include_router(file_locks_router)
app.include_router(jupyter_router)
app.include_router(resources_router)
app.include_router(schedule_runs_router)
backend/src/backend/runtime_client.py
+27 -76
View File
@@ -1,25 +1,32 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import httpx
@dataclass(frozen=True)
class RuntimeClientError(Exception):
status_code: int
detail: Any
from common.clients.base import BaseInternalClient, InternalClientError
class RuntimeClient:
def __init__(
self,
client: httpx.AsyncClient,
service_token: str,
) -> None:
self.client = client
self.headers = {"X-Service-Token": service_token}
class RuntimeClientError(InternalClientError):
"""Backward-compatible alias for the runtime error type."""
_RUNTIME_TRANSPORT_ERROR = RuntimeClientError(
503,
{
"code": "RUNTIME_UNAVAILABLE",
"message": "Runtime Manager 暂时不可用",
"retryable": True,
"details": {},
},
)
class RuntimeClient(BaseInternalClient):
error_class = RuntimeClientError
def __init__(self, client: httpx.AsyncClient) -> None:
super().__init__(client)
async def _request(
self,
@@ -27,68 +34,12 @@ class RuntimeClient:
path: str,
payload: dict[str, Any],
) -> dict[str, Any]:
try:
response = await self.client.request(
method,
path,
json=payload,
headers=self.headers,
)
except httpx.RequestError as exc:
raise RuntimeClientError(
503,
{
"code": "RUNTIME_UNAVAILABLE",
"message": "Runtime Manager 暂时不可用",
"retryable": True,
"details": {},
},
) from exc
if response.is_error:
try:
detail = response.json().get("detail", response.text)
except ValueError:
detail = response.text
raise RuntimeClientError(response.status_code, detail)
return response.json()
async def acquire_file_lock(
self,
payload: dict[str, Any],
) -> dict[str, Any]:
return (
await self._request(
"POST",
"/internal/v1/file-locks/acquire",
payload,
)
)["data"]
async def heartbeat_file_lock(
self,
edit_session_id: str,
payload: dict[str, Any],
) -> dict[str, Any]:
return (
await self._request(
"POST",
f"/internal/v1/file-locks/{edit_session_id}/heartbeat",
payload,
)
)["data"]
async def release_file_lock(
self,
edit_session_id: str,
payload: dict[str, Any],
) -> dict[str, Any]:
return (
await self._request(
"DELETE",
f"/internal/v1/file-locks/{edit_session_id}",
payload,
)
)["data"]
return await super()._request(
method,
path,
payload=payload,
on_transport_error=_RUNTIME_TRANSPORT_ERROR,
)
async def get_workspace(
self,
backend/src/backend/schedule_runs.py
+1 -1
View File
@@ -20,7 +20,7 @@ from backend.dependencies import (
database_session,
request_context,
)
from backend.schedule_schemas import StrictModel
from common.schemas import StrictModel
from backend.schedules import (
graph_rows,
schedule_row,
backend/src/backend/schedule_schemas.py
+3 -6
View File
@@ -1,15 +1,12 @@
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from typing import Any
from pydantic import Field, field_validator, model_validator
from backend.schemas import StrictModel
TriggerType = Literal["manual", "cron", "api"]
FailurePolicy = Literal["stop", "continue"]
from common.db.models.schedules import FailurePolicy, TriggerType
from common.schemas import StrictModel
def _required_text(value: str) -> str:
backend/src/backend/schemas.py
+2 -8
View File
@@ -2,11 +2,9 @@ from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import Field, field_validator
class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid")
from common.schemas import StrictModel
class CreateResourceUploadRequest(StrictModel):
@@ -61,7 +59,3 @@ class PublishVersionRequest(StrictModel):
class DownloadUrlRequest(StrictModel):
expires_seconds: int = Field(default=300, ge=30, le=3600)
class FileLockTokenRequest(StrictModel):
lock_token: str = Field(min_length=32, max_length=256)
backend/src/backend/storage_api.py
+73 -162
View File
@@ -12,28 +12,26 @@ from datetime import UTC, datetime, timedelta
from pathlib import Path, PurePosixPath
from typing import Any, AsyncIterator
from fastapi import Depends, Header, HTTPException, Request, status
from fastapi import Depends, HTTPException, Request, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from common.db import create_database_engine, create_session_factory
from common.db import create_database_engine, create_session_factory, session_scope
from common.db.models import (
StorageObjects,
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 backend.storage_schemas import (
from common.storage.schemas import (
CompleteUploadRequest,
CreateUploadRequest,
DownloadUrlRequest,
RegisterWorkspaceObjectRequest,
ServerObjectRequest,
)
ServerObjectRequest)
def utcnow() -> datetime:
@@ -47,8 +45,7 @@ def hash_bytes(value: str) -> bytes:
def normalized_idempotency_key(
workspace_id: str,
user_id: str,
value: str,
) -> str:
value: str) -> str:
digest = hashlib.sha256(
f"{workspace_id}:{user_id}:{value}".encode("utf-8")
).hexdigest()
@@ -92,23 +89,18 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
app.state.object_store = RustFSObjectStore(
internal_endpoint=os.getenv(
"RUSTFS_INTERNAL_ENDPOINT",
"http://rustfs:9000",
),
"http://rustfs:9000"),
public_endpoint=os.getenv(
"RUSTFS_PUBLIC_ENDPOINT",
"http://localhost:9000",
),
"http://localhost:9000"),
access_key=os.environ["RUSTFS_ACCESS_KEY"],
secret_key=os.environ["RUSTFS_SECRET_KEY"],
)
secret_key=os.environ["RUSTFS_SECRET_KEY"])
app.state.default_bucket = os.getenv(
"RUSTFS_DEFAULT_BUCKET",
"model-platform",
)
"model-platform")
await asyncio.to_thread(
app.state.object_store.ensure_bucket,
app.state.default_bucket,
)
app.state.default_bucket)
try:
yield
finally:
@@ -117,75 +109,51 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
app = create_service_app(
os.getenv("SERVICE_NAME", "storage-api"),
lifespan=lifespan,
)
lifespan=lifespan)
async def database_session(request: Request) -> AsyncIterator[AsyncSession]:
async with request.app.state.session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
def verify_internal_service(
x_service_token: str = Header(alias="X-Service-Token"),
) -> None:
expected = os.environ.get("INTERNAL_SERVICE_TOKEN", "")
if not expected or not secrets.compare_digest(expected, x_service_token):
raise HTTPException(
status.HTTP_401_UNAUTHORIZED,
"invalid internal service identity",
)
async with session_scope(request.app.state.session_factory) as session:
yield session
async def require_workspace_member(
session: AsyncSession,
workspace_id: str,
user_id: str,
) -> Workspaces:
user_id: str) -> Workspaces:
statement = (
select(Workspaces)
.join(
WorkspaceMembers,
WorkspaceMembers.workspace_id == Workspaces.workspace_id,
)
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",
)
"user is not an active workspace member")
return workspace
async def create_upload_record(
payload: CreateUploadRequest,
session: AsyncSession,
request: Request,
) -> dict[str, Any]:
request: Request) -> dict[str, Any]:
workspace = await require_workspace_member(
session,
payload.workspace_id,
payload.user_id,
)
payload.user_id)
stored_key = normalized_idempotency_key(
payload.workspace_id,
payload.user_id,
payload.idempotency_key,
)
payload.idempotency_key)
existing = await session.scalar(
select(UploadSessions).where(
UploadSessions.idempotency_key == stored_key
@@ -201,8 +169,7 @@ 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()
@@ -226,16 +193,14 @@ 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)
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,
)
upload.storage_object_id)
return {
"upload_id": upload.upload_id,
"status": upload.upload_status,
@@ -246,8 +211,7 @@ 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,
@@ -255,8 +219,7 @@ async def create_upload_record(
content_type=upload.content_type or "application/octet-stream",
expected_hash=upload.expected_hash,
expires_seconds=900,
public=payload.url_scope == "public",
)
public=payload.url_scope == "public")
return {
"upload_id": upload.upload_id,
"status": upload.upload_status,
@@ -271,8 +234,7 @@ async def complete_upload_record(
upload_id: str,
payload: CompleteUploadRequest,
session: AsyncSession,
request: Request,
) -> StorageObjects:
request: Request) -> StorageObjects:
upload = await session.scalar(
select(UploadSessions)
.where(UploadSessions.upload_id == upload_id)
@@ -285,14 +247,12 @@ async def complete_upload_record(
if item is None:
raise HTTPException(
status.HTTP_409_CONFLICT,
"completed upload has no storage object",
)
"completed upload has no storage object")
return item
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 be completed from status {upload.upload_status}")
if upload.expires_at < utcnow():
upload.upload_status = "expired"
raise HTTPException(status.HTTP_409_CONFLICT, "upload expired")
@@ -301,13 +261,11 @@ async def complete_upload_record(
head = await asyncio.to_thread(
request.app.state.object_store.head,
bucket_name=upload.bucket_name,
object_key=upload.object_key,
)
object_key=upload.object_key)
except Exception as exc:
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded object is not available",
) from exc
"uploaded object is not available") from exc
actual_size = int(head.get("ContentLength", 0))
if (
@@ -317,8 +275,7 @@ async def complete_upload_record(
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded object size does not match expected_size_bytes",
)
"uploaded object size does not match expected_size_bytes")
actual_content_type = str(
head.get("ContentType") or "application/octet-stream"
)
@@ -326,8 +283,7 @@ async def complete_upload_record(
upload.upload_status = "failed"
raise HTTPException(
status.HTTP_409_CONFLICT,
"uploaded object content type does not match",
)
"uploaded object content type does not match")
metadata = {
str(key).lower(): str(value).lower()
for key, value in dict(head.get("Metadata") or {}).items()
@@ -337,14 +293,12 @@ async def complete_upload_record(
actual_hash = await asyncio.to_thread(
request.app.state.object_store.sha256,
bucket_name=upload.bucket_name,
object_key=upload.object_key,
)
object_key=upload.object_key)
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",
)
"uploaded object hash does not match expected_hash")
file_name = upload.object_key.rsplit("/", 1)[-1]
item = StorageObjects(
@@ -367,8 +321,7 @@ async def complete_upload_record(
visibility=payload.visibility,
is_immutable=int(payload.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)
@@ -378,45 +331,37 @@ async def complete_upload_record(
return item
@app.post("/internal/v1/uploads", dependencies=[Depends(verify_internal_service)])
@app.post("/internal/v1/uploads")
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",
dependencies=[Depends(verify_internal_service)],
)
"/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]:
session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
item = await complete_upload_record(
upload_id,
payload,
session,
request,
)
request)
return {"data": storage_payload(item)}
@app.post(
"/internal/v1/uploads/{upload_id}/abort",
dependencies=[Depends(verify_internal_service)],
)
"/internal/v1/uploads/{upload_id}/abort")
async def abort_upload(
upload_id: str,
request: Request,
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
upload = await session.scalar(
select(UploadSessions)
.where(UploadSessions.upload_id == upload_id)
@@ -427,39 +372,32 @@ async def abort_upload(
if upload.upload_status == "completed":
raise HTTPException(
status.HTTP_409_CONFLICT,
"completed upload cannot be aborted",
)
"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,
)
object_key=upload.object_key)
upload.upload_status = "aborted"
return {"data": {"upload_id": upload_id, "status": "aborted"}}
@app.post(
"/internal/v1/objects",
dependencies=[Depends(verify_internal_service)],
)
"/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]:
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
"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",
)
"object exceeds 100 MiB server-side upload limit")
content_hash = hashlib.sha256(content).hexdigest()
upload_result = await create_upload_record(
CreateUploadRequest(
@@ -471,11 +409,9 @@ async def create_server_object(
expected_size_bytes=len(content),
expected_hash=content_hash,
idempotency_key=payload.idempotency_key,
url_scope="internal",
),
url_scope="internal"),
session,
request,
)
request)
if upload_result.get("status") == "completed":
return {"data": upload_result["storage_object"], "meta": {"reused": True}}
@@ -483,42 +419,34 @@ async def create_server_object(
if upload is None:
raise HTTPException(
status.HTTP_500_INTERNAL_SERVER_ERROR,
"upload record disappeared",
)
"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,
)
content_hash=content_hash)
item = await complete_upload_record(
upload.upload_id,
CompleteUploadRequest(
usage_type=payload.usage_type,
visibility=payload.visibility,
is_immutable=payload.is_immutable,
),
is_immutable=payload.is_immutable),
session,
request,
)
request)
return {"data": storage_payload(item), "meta": {"reused": False}}
@app.post(
"/internal/v1/workspace-objects",
dependencies=[Depends(verify_internal_service)],
)
"/internal/v1/workspace-objects")
async def register_workspace_object(
payload: RegisterWorkspaceObjectRequest,
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
session: AsyncSession = Depends(database_session)) -> dict[str, Any]:
workspace = await require_workspace_member(
session,
payload.workspace_id,
payload.user_id,
)
payload.user_id)
pure_path = PurePosixPath(payload.relative_path.replace("\\", "/"))
if (
pure_path.is_absolute()
@@ -527,8 +455,7 @@ async def register_workspace_object(
):
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
"invalid workspace relative_path",
)
"invalid workspace relative_path")
relative_path = pure_path.as_posix()
workspace_root = Path(
os.getenv("WORKSPACE_ROOT", "/workspace/workspaces")
@@ -538,13 +465,11 @@ async def register_workspace_object(
if scoped_root != target and scoped_root not in target.parents:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
"workspace path escapes its root",
)
"workspace path escapes its root")
if not target.is_file():
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"workspace file does not exist",
)
"workspace file does not exist")
content = await asyncio.to_thread(target.read_bytes)
content_hash = hashlib.sha256(content).hexdigest()
stat_result = target.stat()
@@ -553,8 +478,7 @@ async def register_workspace_object(
select(StorageObjects).where(
StorageObjects.workspace_id == payload.workspace_id,
StorageObjects.storage_backend == "workspace_fs",
StorageObjects.path_hash == path_digest,
)
StorageObjects.path_hash == path_digest)
)
reused = item is not None
if item is None:
@@ -572,19 +496,16 @@ async def register_workspace_object(
visibility=payload.visibility,
is_immutable=0,
object_status="available",
created_by=payload.user_id,
)
created_by=payload.user_id)
session.add(item)
elif item.is_immutable:
raise HTTPException(
status.HTTP_409_CONFLICT,
"immutable workspace object cannot be updated",
)
"immutable workspace object cannot be updated")
elif item.owner_user_id != payload.user_id:
raise HTTPException(
status.HTTP_403_FORBIDDEN,
"workspace object belongs to another user",
)
"workspace object belongs to another user")
item.usage_type = payload.usage_type
item.file_name = target.name
item.file_extension = target.suffix.lower() or None
@@ -601,15 +522,12 @@ async def register_workspace_object(
@app.post(
"/internal/v1/objects/{storage_object_id}/download-url",
dependencies=[Depends(verify_internal_service)],
)
"/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")
@@ -620,14 +538,12 @@ async def create_download_url(
):
raise HTTPException(
status.HTTP_409_CONFLICT,
"object does not support a presigned URL",
)
"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,
)
expires_seconds=payload.expires_seconds)
return {
"data": {
"storage_object_id": item.storage_object_id,
@@ -639,14 +555,11 @@ async def create_download_url(
@app.delete(
"/internal/v1/objects/{storage_object_id}",
dependencies=[Depends(verify_internal_service)],
)
"/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]:
item = await session.scalar(
select(StorageObjects)
.where(StorageObjects.storage_object_id == storage_object_id)
@@ -657,15 +570,13 @@ async def delete_object(
if item.is_immutable:
raise HTTPException(
status.HTTP_409_CONFLICT,
"immutable object cannot be deleted",
)
"immutable object cannot be deleted")
if item.object_status != "deleted":
if item.storage_backend == "rustfs" and item.bucket_name and item.object_key:
await asyncio.to_thread(
request.app.state.object_store.delete,
bucket_name=item.bucket_name,
object_key=item.object_key,
)
object_key=item.object_key)
item.object_status = "deleted"
item.deleted_at = utcnow()
return {
backend/src/backend/storage_client.py
+3 -114
View File
@@ -1,116 +1,5 @@
from __future__ import annotations
"""Backward-compatible re-export of the shared storage client."""
import base64
from typing import Any
from common.storage.client import StorageClient
import httpx
from fastapi import HTTPException
class StorageClient:
def __init__(
self,
client: httpx.AsyncClient,
service_token: str,
) -> None:
self.client = client
self.headers = {"X-Service-Token": service_token}
async def _request(
self,
method: str,
path: str,
*,
payload: dict[str, Any] | None = None,
) -> dict[str, Any]:
response = await self.client.request(
method,
path,
json=payload,
headers=self.headers,
)
if response.is_error:
try:
detail = response.json().get("detail", response.text)
except ValueError:
detail = response.text
raise HTTPException(response.status_code, detail)
return response.json()
async def create_upload(
self,
payload: dict[str, Any],
) -> dict[str, Any]:
return (await self._request(
"POST",
"/internal/v1/uploads",
payload=payload,
))["data"]
async def complete_upload(
self,
upload_id: str,
payload: dict[str, Any],
) -> dict[str, Any]:
return (await self._request(
"POST",
f"/internal/v1/uploads/{upload_id}/complete",
payload=payload,
))["data"]
async def register_workspace_object(
self,
payload: dict[str, Any],
) -> dict[str, Any]:
return (await self._request(
"POST",
"/internal/v1/workspace-objects",
payload=payload,
))["data"]
async def create_server_object(
self,
*,
workspace_id: str,
user_id: str,
usage_type: str,
file_name: str,
content_type: str,
content: bytes,
visibility: str,
is_immutable: bool,
idempotency_key: str,
) -> dict[str, Any]:
result = await self._request(
"POST",
"/internal/v1/objects",
payload={
"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": visibility,
"is_immutable": is_immutable,
"idempotency_key": idempotency_key,
},
)
return result["data"]
async def create_download_url(
self,
storage_object_id: str,
expires_seconds: int,
) -> dict[str, Any]:
return (await self._request(
"POST",
f"/internal/v1/objects/{storage_object_id}/download-url",
payload={"expires_seconds": expires_seconds},
))["data"]
async def delete_object(self, storage_object_id: str) -> dict[str, Any]:
return (await self._request(
"DELETE",
f"/internal/v1/objects/{storage_object_id}",
))["data"]
__all__ = ["StorageClient"]
backend/src/backend/storage_schemas.py
-79
View File
@@ -1,79 +0,0 @@
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid")
class CreateUploadRequest(StrictModel):
workspace_id: str = Field(min_length=26, max_length=26)
user_id: str = Field(min_length=26, max_length=26)
usage_type: Literal[
"data_resource",
"version_artifact",
"snapshot",
"run_log",
"run_result",
]
file_name: str = Field(min_length=1, max_length=255)
content_type: str = Field(min_length=1, max_length=255)
expected_size_bytes: int = Field(ge=0, le=100 * 1024 * 1024)
expected_hash: str | None = Field(default=None, min_length=64, max_length=64)
idempotency_key: str = Field(min_length=8, max_length=128)
url_scope: Literal["public", "internal"] = "public"
@field_validator("expected_hash")
@classmethod
def validate_hash(cls, value: str | None) -> str | None:
if value is None:
return None
normalized = value.lower()
if any(character not in "0123456789abcdef" for character in normalized):
raise ValueError("expected_hash must be lowercase SHA-256 hex")
return normalized
class CompleteUploadRequest(StrictModel):
usage_type: Literal[
"data_resource",
"version_artifact",
"snapshot",
"run_log",
"run_result",
]
visibility: Literal["private", "workspace", "public"] = "private"
is_immutable: bool = False
class ServerObjectRequest(StrictModel):
workspace_id: str = Field(min_length=26, max_length=26)
user_id: str = Field(min_length=26, max_length=26)
usage_type: Literal[
"data_resource",
"version_artifact",
"snapshot",
"run_log",
"run_result",
]
file_name: str = Field(min_length=1, max_length=255)
content_type: str = Field(min_length=1, max_length=255)
content_base64: str = Field(min_length=1)
visibility: Literal["private", "workspace", "public"] = "private"
is_immutable: bool = False
idempotency_key: str = Field(min_length=8, max_length=128)
class RegisterWorkspaceObjectRequest(StrictModel):
workspace_id: str = Field(min_length=26, max_length=26)
user_id: str = Field(min_length=26, max_length=26)
relative_path: str = Field(min_length=1, max_length=1024)
usage_type: Literal["working_copy", "public_script"]
visibility: Literal["private", "workspace", "public"] = "private"
class DownloadUrlRequest(StrictModel):
expires_seconds: int = Field(default=300, ge=30, le=3600)