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

fix: file lock

Browse Source
This commit is contained in:
tao.chen
2026-08-11 21:26:55 +08:00
parent d4013a07a9
commit dd6f176ca8
13 changed files with 345 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/backend/schemas.py
+5 -4
View File
@@ -1,10 +1,7 @@
from __future__ import annotations
from typing import Literal
from pydantic import Field, field_validator
from common.schemas import StrictModel
from pydantic import Field, field_validator
class CreateResourceUploadRequest(StrictModel):
@@ -47,6 +44,10 @@ class UpdateScriptRequest(StrictModel):
content: str = Field(max_length=10 * 1024 * 1024)
class LockScriptRequest(StrictModel):
is_locked: bool
class PublishVersionRequest(StrictModel):
source_object_id: str | None = Field(
default=None,
backend/src/backend/scripts.py
+139 -65
View File
@@ -1,5 +1,3 @@
from __future__ import annotations
import asyncio
import base64
import hashlib
@@ -8,7 +6,15 @@ import mimetypes
from datetime import UTC, datetime
from pathlib import PurePosixPath
from typing import Any
from loguru import logger
from common.config import settings
from common.db.models import (
Scripts,
StorageObjects,
Versions,
)
from common.ids import new_ulid
from common.storage.schemas import ServerObjectRequest
from fastapi import (
APIRouter,
BackgroundTasks,
@@ -19,34 +25,29 @@ from fastapi import (
Request,
status,
)
from loguru import logger
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from common.config import settings
from common.db.models import (
Scripts,
StorageObjects,
Versions,
)
from common.ids import new_ulid
from backend.dependencies import (
RequestContext,
database_session,
request_context,
)
from backend.runtime_client import RuntimeClientError
from backend.services.storage import (
create_download_url_payload,
create_server_object_payload,
)
from common.storage.schemas import ServerObjectRequest
from backend.schemas import (
CreateScriptRequest,
CreateWorkspaceDirectoryRequest,
DownloadUrlRequest,
LockScriptRequest,
PublishVersionRequest,
UpdateScriptRequest,
)
from backend.services.storage import (
create_download_url_payload,
create_server_object_payload,
)
router = APIRouter(tags=["scripts"])
@@ -170,7 +171,7 @@ def script_payload(
size_bytes = storage_object.size_bytes
workspace_prefix = f"{script.workspace_id}/"
jupyter_path = (
object_key[len(workspace_prefix):]
object_key[len(workspace_prefix) :]
if object_key and object_key.startswith(workspace_prefix)
else object_key
)
@@ -183,6 +184,7 @@ def script_payload(
"script_type": script.script_type,
"visibility": script.visibility,
"status": script.status,
"is_locked": bool(script.is_locked),
"relative_path": relative_path,
"jupyter_path": jupyter_path,
"content_hash": content_hash,
@@ -249,8 +251,7 @@ async def get_script_row(
select(Scripts, StorageObjects)
.join(
StorageObjects,
StorageObjects.storage_object_id
== Scripts.current_object_id,
StorageObjects.storage_object_id == Scripts.current_object_id,
)
.where(
Scripts.script_id == script_id,
@@ -304,8 +305,10 @@ async def create_script_record(
session: AsyncSession,
) -> tuple[Scripts, dict[str, Any]]:
folder = (
"scripts" if script_type == "python" else "notebooks"
) if parent_path is None else normalize_user_path(parent_path)
("scripts" if script_type == "python" else "notebooks")
if parent_path is None
else normalize_user_path(parent_path)
)
child_path = f"{folder}/{name}" if folder else name
relative_path = user_relative_path(context, child_path)
logger.debug(relative_path)
@@ -352,9 +355,7 @@ async def create_script_record(
workspace_id,
name=jupyter_name,
content=content.decode("utf-8"),
content_type=(
mimetypes.guess_type(jupyter_name)[0] or "text/plain"
),
content_type=(mimetypes.guess_type(jupyter_name)[0] or "text/plain"),
)
logger.debug(jupyter_resp)
except RuntimeClientError as exc:
@@ -512,10 +513,11 @@ async def get_workspace_tree(
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
# Workspace object storage uses implicit directories (object key prefixes),
# so we derive the tree from ``StorageObjects.relative_path`` rather than
# walking a local filesystem. Only paths that start with the user's
# scoped prefix (and that are currently active) contribute.
# Workspace object storage uses implicit directories (object key prefixes)
# plus explicit directory rows, so we derive the tree from
# ``StorageObjects.relative_path`` and ``StorageObjects.object_type``
# rather than walking a local filesystem. Only paths that start with the
# user's scoped prefix (and that are currently active) contribute.
scoped_prefix = user_relative_path(context)
if scoped_prefix:
like_prefix = f"{scoped_prefix}%"
@@ -524,17 +526,16 @@ async def get_workspace_tree(
rows = (
await session.execute(
select(StorageObjects.relative_path)
.where(
select(StorageObjects.relative_path, StorageObjects.object_type).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.relative_path.like(like_prefix),
)
)
).scalars().all()
).all()
directories: dict[str, dict[str, str]] = {}
for relative in rows:
for relative, object_type in rows:
if not relative:
continue
# Strip the scoped prefix so the returned paths are workspace-local.
@@ -544,7 +545,19 @@ async def get_workspace_tree(
continue
else:
trimmed = relative
# Materialise every ancestor directory of the file.
# Explicit directory rows are included directly, then we still
# materialise every ancestor directory.
if object_type == "directory" and trimmed:
directories.setdefault(
trimmed,
{
"path": trimmed,
"name": trimmed.rsplit("/", 1)[-1],
"parent_path": ""
if "/" not in trimmed
else trimmed.rsplit("/", 1)[0],
},
)
parts = trimmed.split("/")[:-1]
for index in range(1, len(parts) + 1):
directory_path = "/".join(parts[:index])
@@ -579,13 +592,16 @@ async def create_workspace_directory(
relative_path = user_relative_path(context, child_path)
scoped_prefix = user_relative_path(context)
# Validate parent exists: there must be at least one StorageObject whose
# relative_path is exactly the parent directory (or its prefix).
# relative_path is exactly the parent directory (the directory row itself)
# OR lives somewhere below the parent (any file/dir nested under it).
if parent:
parent_relative = f"{scoped_prefix}/{parent}"
existing_parent = await session.scalar(
select(StorageObjects.storage_object_id).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.relative_path.like(f"{scoped_prefix}%"),
(StorageObjects.relative_path == parent_relative)
| StorageObjects.relative_path.like(f"{parent_relative}/%"),
)
)
if existing_parent is None:
@@ -607,9 +623,32 @@ async def create_workspace_directory(
status.HTTP_409_CONFLICT,
"a file or directory with the same path already exists",
)
directory = StorageObjects(
storage_object_id=new_ulid(),
workspace_id=context.workspace.workspace_id,
object_type="directory",
usage_type="working_copy",
storage_backend="rustfs",
storage_uri=f"inline://directory/{relative_path}",
relative_path=relative_path,
path_hash=hashlib.sha256(relative_path.encode("utf-8")).digest(),
object_status="available",
size_bytes=0,
visibility="private",
created_by=context.user.user_id,
)
session.add(directory)
try:
await session.flush()
except IntegrityError as exc:
raise HTTPException(
status.HTTP_409_CONFLICT,
"directory already exists",
) from exc
return {
"request_id": context.request_id,
"data": {
"storage_object_id": directory.storage_object_id,
"path": child_path,
"name": name,
"parent_path": parent,
@@ -636,15 +675,18 @@ async def delete_workspace_directory(
# to the runtime. ``directory_path`` is still returned in the
# response for API compatibility.
rows = (
await session.execute(
select(Scripts)
.where(
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.owner_user_id == context.user.user_id,
Scripts.status == "active",
(
await session.execute(
select(Scripts).where(
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.owner_user_id == context.user.user_id,
Scripts.status == "active",
)
)
)
).scalars().all()
.scalars()
.all()
)
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
for script in rows:
@@ -700,8 +742,7 @@ async def list_scripts(
return {
"request_id": context.request_id,
"data": [
script_payload(script, storage_object)
for script, storage_object in rows
script_payload(script, storage_object) for script, storage_object in rows
],
"meta": {"count": len(rows)},
}
@@ -773,10 +814,7 @@ async def update_script(
workspace_id,
name=jupyter_name,
content=content.decode("utf-8"),
content_type=(
mimetypes.guess_type(jupyter_name)[0]
or "text/plain"
),
content_type=(mimetypes.guess_type(jupyter_name)[0] or "text/plain"),
)
except RuntimeClientError as exc:
raise HTTPException(
@@ -799,6 +837,48 @@ async def update_script(
}
@router.patch("/api/v1/scripts/{script_id}/lock")
async def set_script_lock(
script_id: str,
payload: LockScriptRequest,
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
script = await session.scalar(
select(Scripts)
.where(
Scripts.script_id == script_id,
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
)
.with_for_update()
)
if script is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"script not found",
)
require_script_modify_access(
script,
user_id=context.user.user_id,
is_admin=context.is_admin,
)
script.is_locked = 1 if payload.is_locked else 0
await session.commit()
storage_object = await session.scalar(
select(StorageObjects).where(
StorageObjects.storage_object_id == script.current_object_id
)
)
return {
"request_id": context.request_id,
"data": script_payload(script, storage_object)
if storage_object
else script_payload(script, {}),
"meta": {},
}
@router.delete("/api/v1/scripts/{script_id}")
async def delete_script(
script_id: str,
@@ -905,18 +985,16 @@ async def publish_version(
workspace_id = context.workspace.workspace_id
jupyter_name = _jupyter_path(script.script_type, script.script_id)
try:
contents = await runtime_client.get_file(
workspace_id, name=jupyter_name
)
contents = await runtime_client.get_file(workspace_id, name=jupyter_name)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
detail=exc.detail,
) from exc
if contents.get("type") == "notebook":
content = json.dumps(
contents.get("content", {}), ensure_ascii=False
).encode("utf-8")
content = json.dumps(contents.get("content", {}), ensure_ascii=False).encode(
"utf-8"
)
else:
content = (contents.get("content") or "").encode("utf-8")
@@ -934,8 +1012,7 @@ async def publish_version(
"meta": {"reused": True},
}
content_type = (
mimetypes.guess_type(script.script_name)[0]
or "application/octet-stream"
mimetypes.guess_type(script.script_name)[0] or "application/octet-stream"
)
artifact = await create_server_object_payload(
ServerObjectRequest(
@@ -1034,9 +1111,12 @@ async def latest_version(
"script not found",
)
latest = await session.scalar(
select(Versions).where(
select(Versions)
.where(
Versions.script_id == script_id,
).order_by(Versions.version_no.desc()).limit(1)
)
.order_by(Versions.version_no.desc())
.limit(1)
)
if latest is None:
return {
@@ -1061,10 +1141,7 @@ async def get_version(
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
version = await session.get(Versions, versions_id)
if (
version is None
or version.workspace_id != context.workspace.workspace_id
):
if version is None or version.workspace_id != context.workspace.workspace_id:
raise HTTPException(status.HTTP_404_NOT_FOUND, "version not found")
return {
"request_id": context.request_id,
@@ -1129,10 +1206,7 @@ async def version_download_url(
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
version = await session.get(Versions, versions_id)
if (
version is None
or version.workspace_id != context.workspace.workspace_id
):
if version is None or version.workspace_id != context.workspace.workspace_id:
raise HTTPException(status.HTTP_404_NOT_FOUND, "version not found")
data = await create_download_url_payload(
await session.get(StorageObjects, version.artifact_object_id),