diff --git a/.env.example b/.env.example index aae7d7f..15cade1 100644 --- a/.env.example +++ b/.env.example @@ -1,33 +1,17 @@ COMPOSE_PROJECT_NAME=model-platform-refactored -# Local development ports -NGINX_PORT=8080 -GATEWAY_PORT=8081 -MYSQL_PORT=3308 -REDIS_PORT=6380 -RUSTFS_API_PORT=9010 -RUSTFS_CONSOLE_PORT=9011 +# ports BACKEND_PORT=8010 RUNTIME_PORT=8012 SCHEDULE_PORT=8013 -# Jupyter runs on the internal Compose network only in step 14. -JUPYTER_IMAGE=quay.io/jupyter/base-notebook:2025-12-31 -JUPYTER_TOKEN=ChangeMe_Jupyter_Internal_2026 +# MySQL connection URI (async SQLAlchemy driver) +DATABASE_URL=mysql+asyncmy://model_platform:ChangeMe_MySQL_App_2026@mysql:3306/model_platform?charset=utf8mb4 -# MySQL 8 local development credentials -MYSQL_DATABASE=model_platform -MYSQL_USER=model_platform -MYSQL_PASSWORD=ChangeMe_MySQL_App_2026 -MYSQL_ROOT_PASSWORD=ChangeMe_MySQL_Root_2026 - -# Redis local development credential -REDIS_PASSWORD=ChangeMe_Redis_2026 - -# RustFS local development image and credentials -RUSTFS_IMAGE=rustfs/rustfs:latest +# Object storage (S3-compatible) endpoints +RUSTFS_INTERNAL_ENDPOINT=http://rustfs:9000 +RUSTFS_PUBLIC_ENDPOINT=http://localhost:9010 RUSTFS_ACCESS_KEY=modelplatform RUSTFS_SECRET_KEY=ChangeMe_RustFS_2026 - -# Internal service authentication; replace in every non-local environment. -INTERNAL_SERVICE_TOKEN=ChangeMe_Internal_Service_2026 +RUSTFS_API_PORT=9010 +RUSTFS_CONSOLE_PORT=9011 diff --git a/backend/src/backend/dependencies.py b/backend/src/backend/dependencies.py index f646a77..615a915 100644 --- a/backend/src/backend/dependencies.py +++ b/backend/src/backend/dependencies.py @@ -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( diff --git a/backend/src/backend/file_locks.py b/backend/src/backend/file_locks.py deleted file mode 100644 index 7a82cb3..0000000 --- a/backend/src/backend/file_locks.py +++ /dev/null @@ -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, - }, - ), - ) diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 5991bac..6d7f83a 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -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) diff --git a/backend/src/backend/runtime_client.py b/backend/src/backend/runtime_client.py index 9e861f1..037e4f4 100644 --- a/backend/src/backend/runtime_client.py +++ b/backend/src/backend/runtime_client.py @@ -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, diff --git a/backend/src/backend/schedule_runs.py b/backend/src/backend/schedule_runs.py index 3ac862b..5147c70 100644 --- a/backend/src/backend/schedule_runs.py +++ b/backend/src/backend/schedule_runs.py @@ -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, diff --git a/backend/src/backend/schedule_schemas.py b/backend/src/backend/schedule_schemas.py index 8d658bb..31c7fbc 100644 --- a/backend/src/backend/schedule_schemas.py +++ b/backend/src/backend/schedule_schemas.py @@ -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: diff --git a/backend/src/backend/schemas.py b/backend/src/backend/schemas.py index cc4676b..a17900b 100644 --- a/backend/src/backend/schemas.py +++ b/backend/src/backend/schemas.py @@ -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) diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index bfd0ffc..70e16c3 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -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 { diff --git a/backend/src/backend/storage_client.py b/backend/src/backend/storage_client.py index 8a57826..96f41e4 100644 --- a/backend/src/backend/storage_client.py +++ b/backend/src/backend/storage_client.py @@ -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"] diff --git a/backend/src/backend/storage_schemas.py b/backend/src/backend/storage_schemas.py deleted file mode 100644 index 0e5f36f..0000000 --- a/backend/src/backend/storage_schemas.py +++ /dev/null @@ -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) diff --git a/common/src/common/db/models/schedules.py b/common/src/common/db/models/schedules.py index d0103b6..2254fe0 100644 --- a/common/src/common/db/models/schedules.py +++ b/common/src/common/db/models/schedules.py @@ -1,6 +1,6 @@ import datetime import decimal -from typing import Optional +from typing import Literal, Optional from sqlalchemy import DECIMAL, Index, Integer, JSON, String, Text, text from sqlalchemy.dialects.mysql import BIGINT, CHAR, DATETIME, INTEGER, TINYINT @@ -9,6 +9,10 @@ from sqlalchemy.orm import Mapped, mapped_column from common.db.base import Base +TriggerType = Literal["manual", "cron", "api"] +FailurePolicy = Literal["stop", "continue"] + + class Schedules(Base): __tablename__ = "schedules" __table_args__ = ( diff --git a/common/src/common/storage/__init__.py b/common/src/common/storage/__init__.py index 417f62a..61ffe4b 100644 --- a/common/src/common/storage/__init__.py +++ b/common/src/common/storage/__init__.py @@ -1,3 +1,6 @@ +"""Storage building blocks shared by backend and schedule services.""" + +from common.storage.client import StorageClient from common.storage.rustfs import RustFSObjectStore -__all__ = ["RustFSObjectStore"] +__all__ = ["RustFSObjectStore", "StorageClient"] diff --git a/deploy/data/workspaces/model-dev/users/admin-zhang/.ipynb_checkpoints/test2-checkpoint.py b/deploy/data/workspaces/model-dev/users/admin-zhang/.ipynb_checkpoints/test2-checkpoint.py deleted file mode 100644 index 2dc3e1b..0000000 --- a/deploy/data/workspaces/model-dev/users/admin-zhang/.ipynb_checkpoints/test2-checkpoint.py +++ /dev/null @@ -1,9 +0,0 @@ -"""模型实验开发平台构建脚本。""" - - -def main() -> None: - print("Hello, Model Platform!") - - -if __name__ == "__main__": - main() diff --git a/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/.ipynb_checkpoints/test-checkpoint.ipynb b/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/.ipynb_checkpoints/test-checkpoint.ipynb deleted file mode 100644 index bcf5520..0000000 --- a/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/.ipynb_checkpoints/test-checkpoint.ipynb +++ /dev/null @@ -1,43 +0,0 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "id": "e139eae2", - "metadata": {}, - "source": [ - "# 新建模型实验\\n在这里开始数据探索与模型构建。" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "7a78bce5", - "metadata": {}, - "outputs": [], - "source": [ - "print('Hello, Model Platform!')" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.13.11" - } - }, - "nbformat": 4, - "nbformat_minor": 5 -} diff --git a/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/NotebookDemo.ipynb b/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/NotebookDemo.ipynb deleted file mode 100644 index 7cd3200..0000000 --- a/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/NotebookDemo.ipynb +++ /dev/null @@ -1,67 +0,0 @@ -{ - "cells": [ - { - "cell_type": "code", - "execution_count": 1, - "id": "initial_id", - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": ["hello world\n"] - } - ], - "source": ["print(\"hello world\")"] - }, - { - "cell_type": "code", - "execution_count": 1, - "id": "979c21d2489e134d", - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": ["hello world12323\n"] - } - ], - "source": ["print(\"hello world12323\")"] - }, - { - "cell_type": "code", - "execution_count": 2, - "id": "e068092a-3b51-4399-9c64-c44d58f4973c", - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": ["Python 3.12.13\n"] - } - ], - "source": ["!python --version"] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "1f512d9e-92f7-423e-9176-40fec30dd79c", - "metadata": {}, - "outputs": [], - "source": [] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "name": "python", - "version": "3.12.13" - } - }, - "nbformat": 4, - "nbformat_minor": 5 -} diff --git a/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/test.ipynb b/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/test.ipynb deleted file mode 100644 index 4fc54c1..0000000 --- a/deploy/data/workspaces/model-dev/users/admin-zhang/notebooks/test.ipynb +++ /dev/null @@ -1,59 +0,0 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "id": "e139eae2", - "metadata": {}, - "source": [ - "# 新建模型实验\\n在这里开始数据探索与模型构建。" - ] - }, - { - "cell_type": "code", - "execution_count": 1, - "id": "7a78bce5", - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "111\n" - ] - } - ], - "source": [ - "print('111')" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "42d2764b-5699-4374-afe9-aecda6d2d1cd", - "metadata": {}, - "outputs": [], - "source": [] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.13.11" - } - }, - "nbformat": 4, - "nbformat_minor": 5 -} diff --git a/deploy/data/workspaces/model-dev/users/admin-zhang/test2.py b/deploy/data/workspaces/model-dev/users/admin-zhang/test2.py deleted file mode 100644 index 2dc3e1b..0000000 --- a/deploy/data/workspaces/model-dev/users/admin-zhang/test2.py +++ /dev/null @@ -1,9 +0,0 @@ -"""模型实验开发平台构建脚本。""" - - -def main() -> None: - print("Hello, Model Platform!") - - -if __name__ == "__main__": - main() diff --git a/deploy/demo/NotebookDemo.ipynb b/deploy/demo/NotebookDemo.ipynb deleted file mode 100644 index 7cd3200..0000000 --- a/deploy/demo/NotebookDemo.ipynb +++ /dev/null @@ -1,67 +0,0 @@ -{ - "cells": [ - { - "cell_type": "code", - "execution_count": 1, - "id": "initial_id", - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": ["hello world\n"] - } - ], - "source": ["print(\"hello world\")"] - }, - { - "cell_type": "code", - "execution_count": 1, - "id": "979c21d2489e134d", - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": ["hello world12323\n"] - } - ], - "source": ["print(\"hello world12323\")"] - }, - { - "cell_type": "code", - "execution_count": 2, - "id": "e068092a-3b51-4399-9c64-c44d58f4973c", - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": ["Python 3.12.13\n"] - } - ], - "source": ["!python --version"] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "1f512d9e-92f7-423e-9176-40fec30dd79c", - "metadata": {}, - "outputs": [], - "source": [] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "name": "python", - "version": "3.12.13" - } - }, - "nbformat": 4, - "nbformat_minor": 5 -} diff --git a/docker-compose.legacy.yml b/docker-compose.legacy.yml deleted file mode 100644 index 9d27e38..0000000 --- a/docker-compose.legacy.yml +++ /dev/null @@ -1,60 +0,0 @@ -version: '3.8' - -services: - web: - image: nginx:alpine - ports: - - "8888:80" - restart: unless-stopped - volumes: - - ./default.conf:/etc/nginx/conf.d/default.conf:ro - depends_on: - - backend - - backend: - build: - context: . - dockerfile: backend/Dockerfile - ports: - - "8004:8000" - environment: - - RUNTIME_BASE_URL=http://runtime:8001 - volumes: - - ./backend:/app/backend:ro - - ./common:/app/common:ro - depends_on: - - runtime - - runtime: - build: - context: . - dockerfile: runtime/Dockerfile - cap_add: - - SYS_ADMIN - devices: - - /dev/fuse:/dev/fuse - security_opt: - - apparmor:unconfined - - ports: - - "8002:8001" - environment: - - PUBLIC_BASE_URL=http://runtime - # --- Rclone 动态环境变量配置 (对应名称 rustfs) --- - - RCLONE_CONFIG_RUSTFS_TYPE=s3 - - RCLONE_CONFIG_RUSTFS_PROVIDER=Other - - RCLONE_CONFIG_RUSTFS_ACCESS_KEY_ID=BdsXeamEnvSDQnk8tRxh - - RCLONE_CONFIG_RUSTFS_SECRET_ACCESS_KEY=mmBVc3RqzbT2VX3ysKGnirYH5kYD3ww3wFtMVvrb - # 替换为你的 RustFS 服务地址（如果是同 docker-compose 网络下的服务，可以直接填服务名:端口） - - RCLONE_CONFIG_RUSTFS_ENDPOINT=http://8.153.151.51:9000 - # 自建 S3 建议强制开启 Path-style 访问 (http://endpoint/bucket) - - RCLONE_CONFIG_RUSTFS_ENV_AUTH=false - - RCLONE_CONFIG_RUSTFS_FORCE_PATH_STYLE=true - - RCLONE_CONFIG_RUSTFS_REGION=other - - # --- Runtime 逻辑环境变量 --- - - REMOTE_BUCKET=rustfs:workspaces - - WORKSPACES_ROOT=/app/workspaces - volumes: - - ./runtime:/app/runtime:ro - - ./common:/app/common:ro \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 16fdec1..43df31b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,100 +1,16 @@ -name: ${COMPOSE_PROJECT_NAME:-model-platform-refactored} - -x-app-environment: &app-environment - DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 - INTERNAL_SERVICE_TOKEN: ${INTERNAL_SERVICE_TOKEN:-local-internal-token} - JUPYTER_TOKEN: ${JUPYTER_TOKEN:-local-jupyter-token} - WORKSPACE_ROOT: /workspace/workspaces +version: '3.8' services: - mysql: - image: mysql:8.0.36 + web: + image: nginx:alpine restart: unless-stopped - environment: - MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD:-model_platform_root} - MYSQL_DATABASE: ${MYSQL_DATABASE:-model_platform} - MYSQL_USER: ${MYSQL_USER:-model_platform} - MYSQL_PASSWORD: ${MYSQL_PASSWORD:-model_platform} - command: - - --character-set-server=utf8mb4 - - --collation-server=utf8mb4_0900_ai_ci ports: - - "${MYSQL_PORT:-3308}:3306" + - "${GATEWAY_PORT:-8888}:80" volumes: - - mysql_data:/var/lib/mysql - healthcheck: - test: ["CMD-SHELL", "mysqladmin ping -h 127.0.0.1 -u root -p$$MYSQL_ROOT_PASSWORD --silent"] - interval: 10s - timeout: 5s - retries: 15 - - redis: - image: redis:7.2.5-alpine - restart: unless-stopped - command: ["redis-server", "--appendonly", "yes", "--requirepass", "${REDIS_PASSWORD:-model_platform_redis}"] - ports: - - "${REDIS_PORT:-6380}:6379" - volumes: - - redis_data:/data - healthcheck: - test: ["CMD-SHELL", "redis-cli -a '${REDIS_PASSWORD:-model_platform_redis}' ping | grep PONG"] - interval: 10s - timeout: 5s - retries: 10 - - rustfs: - image: ${RUSTFS_IMAGE:-rustfs/rustfs:latest} - restart: unless-stopped - environment: - RUSTFS_ADDRESS: ":9000" - RUSTFS_CONSOLE_ENABLE: "true" - RUSTFS_CONSOLE_ADDRESS: ":9001" - RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:-modelplatform} - RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} - command: ["/data"] - ports: - - "${RUSTFS_API_PORT:-9010}:9000" - - "${RUSTFS_CONSOLE_PORT:-9011}:9001" - volumes: - - rustfs_data:/data - - jupyter: - image: ${JUPYTER_IMAGE:-quay.io/jupyter/base-notebook:2025-12-31} - restart: unless-stopped - environment: - JUPYTER_TOKEN: ${JUPYTER_TOKEN:-local-jupyter-token} - command: - - start-notebook.py - - --ServerApp.base_url=/jupyter/ - - --ServerApp.root_dir=/home/jovyan/work - - --ServerApp.ip=0.0.0.0 - - --ServerApp.allow_remote_access=True - - --IdentityProvider.token=${JUPYTER_TOKEN:-local-jupyter-token} - - --PasswordIdentityProvider.hashed_password= - volumes: - - ./deploy/data/workspaces:/home/jovyan/work - expose: - - "8888" - - migrate: - build: - context: . - dockerfile: backend/Dockerfile - environment: - <<: *app-environment - command: - - uv - - run - - --frozen - - --package - - backend - - alembic - - upgrade - - head + - ./default.conf:/etc/nginx/conf.d/default.conf:ro depends_on: - mysql: - condition: service_healthy - restart: "no" + - backend + - runtime backend: build: @@ -102,38 +18,39 @@ services: dockerfile: backend/Dockerfile restart: unless-stopped environment: - <<: *app-environment - SERVICE_NAME: backend - READINESS_TARGETS: mysql:3306,redis:6379,rustfs:9000 + DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 + JWT_SECRET: ${JWT_SECRET:-local-jwt-secret} RUNTIME_API_URL: http://runtime:8000 + RUNTIME_BASE_URL: http://runtime:8000 RUSTFS_INTERNAL_ENDPOINT: http://rustfs:9000 RUSTFS_PUBLIC_ENDPOINT: http://localhost:${RUSTFS_API_PORT:-9010} RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:-modelplatform} RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} RUSTFS_DEFAULT_BUCKET: model-platform + WORKSPACE_ROOT: /workspace/workspaces volumes: - ./deploy/data/workspaces:/workspace/workspaces + - ./backend:/app/backend:ro + - ./common:/app/common:ro ports: - - "${BACKEND_PORT:-8010}:8000" + - "${BACKEND_PORT:-8004}:8000" depends_on: - migrate: - condition: service_completed_successfully - mysql: - condition: service_healthy - redis: - condition: service_healthy - rustfs: - condition: service_started + - runtime runtime: build: context: . dockerfile: runtime/Dockerfile restart: unless-stopped + cap_add: + - SYS_ADMIN + devices: + - /dev/fuse:/dev/fuse + security_opt: + - apparmor:unconfined environment: - <<: *app-environment + DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 SERVICE_NAME: runtime-manager - READINESS_TARGETS: mysql:3306,redis:6379,jupyter:8888 REDIS_HOST: redis REDIS_PORT: "6379" REDIS_PASSWORD: ${REDIS_PASSWORD:-model_platform_redis} @@ -141,15 +58,23 @@ services: JUPYTER_INTERNAL_URL: http://jupyter:8888/jupyter/ JUPYTER_PROXY_BASE_PATH: /jupyter/ JUPYTER_TICKET_TTL_SECONDS: "300" + WORKSPACES_ROOT: /workspace/workspaces + PUBLIC_BASE_URL: http://runtime + REMOTE_BUCKET: rustfs:workspaces + RCLONE_CONFIG_RUSTFS_TYPE: s3 + RCLONE_CONFIG_RUSTFS_PROVIDER: Other + RCLONE_CONFIG_RUSTFS_ACCESS_KEY_ID: ${RUSTFS_ACCESS_KEY:-modelplatform} + RCLONE_CONFIG_RUSTFS_SECRET_ACCESS_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} + RCLONE_CONFIG_RUSTFS_ENDPOINT: http://rustfs:9000 + RCLONE_CONFIG_RUSTFS_ENV_AUTH: "false" + RCLONE_CONFIG_RUSTFS_FORCE_PATH_STYLE: "true" + RCLONE_CONFIG_RUSTFS_REGION: other + volumes: + - ./deploy/data/workspaces:/workspace/workspaces + - ./runtime:/app/runtime:ro + - ./common:/app/common:ro ports: - - "${RUNTIME_PORT:-8012}:8000" - depends_on: - mysql: - condition: service_healthy - redis: - condition: service_healthy - jupyter: - condition: service_started + - "${RUNTIME_PORT:-8002}:8001" schedule: build: @@ -157,9 +82,7 @@ services: dockerfile: schedule/Dockerfile restart: unless-stopped environment: - <<: *app-environment - SERVICE_NAME: schedule-executor - READINESS_TARGETS: mysql:3306,redis:6379,rustfs:9000,backend:8000 + DATABASE_URL: mysql+asyncmy://${MYSQL_USER:-model_platform}:${MYSQL_PASSWORD:-model_platform}@mysql:3306/${MYSQL_DATABASE:-model_platform}?charset=utf8mb4 REDIS_HOST: redis REDIS_PORT: "6379" REDIS_PASSWORD: ${REDIS_PASSWORD:-model_platform_redis} @@ -167,36 +90,10 @@ services: RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:-modelplatform} RUSTFS_SECRET_KEY: ${RUSTFS_SECRET_KEY:-modelplatformsecret} STORAGE_API_URL: http://backend:8000 + WORKSPACE_ROOT: /workspace/workspaces volumes: - ./deploy/data/workspaces:/workspace/workspaces ports: - "${SCHEDULE_PORT:-8013}:8000" depends_on: - mysql: - condition: service_healthy - redis: - condition: service_healthy - backend: - condition: service_started - - gateway: - image: nginx:1.27-alpine - restart: unless-stopped - environment: - INTERNAL_SERVICE_TOKEN: ${INTERNAL_SERVICE_TOKEN:-local-internal-token} - volumes: - - ./nginx/default.conf.template:/etc/nginx/templates/default.conf.template:ro - ports: - - "${GATEWAY_PORT:-8081}:80" - depends_on: - backend: - condition: service_started - runtime: - condition: service_started - jupyter: - condition: service_started - -volumes: - mysql_data: - redis_data: - rustfs_data: + - backend diff --git a/schedule/src/schedule/main.py b/schedule/src/schedule/main.py index a5e4478..ece1c3b 100644 --- a/schedule/src/schedule/main.py +++ b/schedule/src/schedule/main.py @@ -26,10 +26,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]: session_factory=session_factory, redis=redis, object_store=build_object_store(), - storage_client=SchedulerStorageClient( - storage_http_client, - os.environ["INTERNAL_SERVICE_TOKEN"], - ), + storage_client=SchedulerStorageClient(storage_http_client), workspace_root=Path( os.getenv("WORKSPACE_ROOT", "/workspace/workspaces") ), diff --git a/schedule/src/schedule/storage_client.py b/schedule/src/schedule/storage_client.py index 3177b5a..11f719d 100644 --- a/schedule/src/schedule/storage_client.py +++ b/schedule/src/schedule/storage_client.py @@ -1,16 +1,13 @@ +"""Schedule-specific storage client built on the shared ``StorageClient``.""" + from __future__ import annotations -import base64 from typing import Any -import httpx +from common.storage.client import StorageClient -class SchedulerStorageClient: - def __init__(self, client: httpx.AsyncClient, service_token: str) -> None: - self.client = client - self.headers = {"X-Service-Token": service_token} - +class SchedulerStorageClient(StorageClient): async def create_object( self, *, @@ -22,20 +19,17 @@ class SchedulerStorageClient: content: bytes, idempotency_key: str, ) -> dict[str, Any]: - response = await self.client.post( - "/internal/v1/objects", - headers=self.headers, - 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, - }, + return await self.create_server_object( + workspace_id=workspace_id, + user_id=user_id, + usage_type=usage_type, + file_name=file_name, + content_type=content_type, + content=content, + visibility="workspace", + is_immutable=True, + idempotency_key=idempotency_key, ) - response.raise_for_status() - return response.json()["data"] + + +__all__ = ["SchedulerStorageClient"]