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

Merge branch 'develop' of http://8.153.151.51:8888/team_group/model-develop into develop

Browse Source
This commit is contained in:
xiaozhu
2026-08-13 10:06:46 +08:00
parent 04447abfb0 da9a1ec472
commit 7b10f08484
17 changed files with 1118 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/backend/scripts.py
+211 -87
View File
@@ -11,10 +11,12 @@ from common.config import settings
from common.db.models import (
Scripts,
StorageObjects,
Users,
Versions,
)
from common.ids import new_ulid
from common.storage.schemas import ServerObjectRequest
from common.storage import build_storage_uri
from fastapi import (
APIRouter,
BackgroundTasks,
@@ -88,8 +90,17 @@ def safe_directory_name(value: str) -> str:
return name
# Version/run artifacts are not part of the user's workspace directory tree.
TREE_EXCLUDED_USAGE_TYPES = (
"version_artifact",
"snapshot",
"run_log",
"run_result",
)
def user_relative_path(context: RequestContext, child_path: str = "") -> str:
base = f"users/{context.user.username}"
base = f"workspace/{context.user.user_id}"
normalized = normalize_user_path(child_path)
return f"{base}/{normalized}" if normalized else base
@@ -123,11 +134,36 @@ def _jupyter_path(script_type: str, script_id: str) -> str:
(``/jupyter/<ws>/notebooks/<file>.ipynb``) is jupyter's URL route
for the editor view, not a filesystem path — jupyter routes that
URL to the file at the workspace root.
NOTE: This function returns the flat basename only (``{script_id}.{ext}``).
It must NOT be used directly as a Jupyter path for scripts that live
inside sub-directories. Callers must combine it with the parent
directory ULID (for newly-created files) or derive the real path from
``StorageObjects.object_key`` (for existing files).
"""
ext = ".ipynb" if script_type == "notebook" else ".py"
return f"{script_id}{ext}"
def _derive_jupyter_path(
storage_object: StorageObjects | None,
workspace_id: str,
script_type: str,
script_id: str,
) -> str:
"""Return the real Jupyter path for an existing script.
For normal workspace files the path is taken from
``StorageObjects.object_key`` with the workspace prefix removed.
For jupyter-only scripts that have no StorageObjects row, fall back
to the flat ``_jupyter_path()`` basename so existing behavior is
preserved.
"""
if storage_object is None or not storage_object.object_key:
return _jupyter_path(script_type, script_id)
return storage_object.object_key.removeprefix(f"{workspace_id}/")
def validate_script_content(content: str, script_type: str) -> bytes:
encoded = content.encode("utf-8")
if len(encoded) > 10 * 1024 * 1024:
@@ -156,9 +192,16 @@ def validate_script_content(content: str, script_type: str) -> bytes:
def script_payload(
script: Scripts,
storage_object: StorageObjects | dict[str, Any],
storage_object: StorageObjects | dict[str, Any] | None,
*,
owner_display_name: str | None = None,
) -> dict[str, Any]:
if isinstance(storage_object, dict):
if storage_object is None:
relative_path = None
object_key = None
content_hash = None
size_bytes = 0
elif isinstance(storage_object, dict):
relative_path = storage_object.get("relative_path")
object_key = storage_object.get("object_key")
content_hash = storage_object.get("content_hash")
@@ -169,16 +212,20 @@ def script_payload(
content_hash = storage_object.content_hash
size_bytes = storage_object.size_bytes
workspace_prefix = f"{script.workspace_id}/"
jupyter_path = (
object_key[len(workspace_prefix) :]
if object_key and object_key.startswith(workspace_prefix)
else object_key
)
if object_key:
jupyter_path = (
object_key[len(workspace_prefix) :]
if object_key.startswith(workspace_prefix)
else object_key
)
else:
jupyter_path = _jupyter_path(script.script_type, script.script_id)
return {
"script_id": script.script_id,
"workspace_id": script.workspace_id,
"current_object_id": script.current_object_id,
"owner_user_id": script.owner_user_id,
"owner_display_name": owner_display_name,
"script_name": script.script_name,
"script_type": script.script_type,
"visibility": script.visibility,
@@ -219,7 +266,8 @@ async def get_script_row(
session: AsyncSession,
*,
for_update: bool = False,
) -> tuple[Scripts, StorageObjects]:
allow_missing_storage_object: bool = False,
) -> tuple[Scripts, StorageObjects | None]:
if for_update:
script = await session.scalar(
select(Scripts)
@@ -239,25 +287,39 @@ async def get_script_row(
StorageObjects,
script.current_object_id,
)
if storage_object is None:
if storage_object is None and not allow_missing_storage_object:
raise HTTPException(
status.HTTP_409_CONFLICT,
"script working-copy metadata is missing",
)
row = (script, storage_object)
else:
statement = (
select(Scripts, StorageObjects)
.join(
StorageObjects,
StorageObjects.storage_object_id == Scripts.current_object_id,
if allow_missing_storage_object:
statement = (
select(Scripts, StorageObjects)
.outerjoin(
StorageObjects,
StorageObjects.storage_object_id == Scripts.current_object_id,
)
.where(
Scripts.script_id == script_id,
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
)
)
.where(
Scripts.script_id == script_id,
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
else:
statement = (
select(Scripts, StorageObjects)
.join(
StorageObjects,
StorageObjects.storage_object_id == Scripts.current_object_id,
)
.where(
Scripts.script_id == script_id,
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
)
)
)
row = (await session.execute(statement)).one_or_none()
if row is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "script not found")
@@ -412,7 +474,7 @@ async def create_script_record(
bucket_name=bucket_name,
object_key=object_key,
object_key_hash=hashlib.sha256(object_key.encode("utf-8")).digest(),
storage_uri=f"s3://{bucket_name}/{object_key}",
storage_uri=build_storage_uri(bucket_name, object_key),
file_name=name,
file_extension=PurePosixPath(jupyter_name).suffix.lower() or None,
mime_type=mime_type,
@@ -571,7 +633,9 @@ async def get_workspace_tree(
select(StorageObjects.relative_path, StorageObjects.object_type).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.is_deleted == 0,
StorageObjects.relative_path.like(like_prefix),
StorageObjects.usage_type.notin_(TREE_EXCLUDED_USAGE_TYPES),
)
)
).all()
@@ -612,6 +676,78 @@ async def get_workspace_tree(
},
)
sorted_dirs = sorted(directories.values(), key=lambda item: item["path"])
return {
"request_id": context.request_id,
"data": {"directories": sorted_dirs},
"meta": {"directory_count": len(sorted_dirs)},
}
@router.get("/api/v1/workspace-directories")
async def list_workspace_directories(
parent_path: str = Query(default=""),
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
"""List direct child directories of a workspace path.
Empty ``parent_path`` returns the directories immediately under the
user's scoped root. Only available, non-deleted StorageObjects are
considered.
"""
scoped_prefix = user_relative_path(context)
parent = normalize_user_path(parent_path)
target_prefix = f"{scoped_prefix}/{parent}" if parent else scoped_prefix
descendant_prefix = f"{target_prefix}/"
rows = (
await session.execute(
select(StorageObjects.relative_path).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.is_deleted == 0,
StorageObjects.relative_path.like(f"{descendant_prefix}%"),
~StorageObjects.relative_path.like(f"{descendant_prefix}%/%"),
StorageObjects.object_type == "directory",
StorageObjects.usage_type.notin_(TREE_EXCLUDED_USAGE_TYPES),
)
)
).all()
directories: dict[str, dict[str, Any]] = {}
for (relative,) in rows:
if not relative or not relative.startswith(descendant_prefix):
continue
suffix = relative[len(descendant_prefix) :]
if "/" in suffix:
continue
child_path = f"{parent}/{suffix}" if parent else suffix
directories.setdefault(
child_path,
{
"path": child_path,
"name": suffix,
"parent_path": parent,
"has_children": False,
},
)
for directory in directories.values():
# directory['path'] is already workspace-relative and includes the parent segment.
child_prefix = f"{scoped_prefix}/{directory['path']}/"
has_children = await session.scalar(
select(StorageObjects.storage_object_id).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.is_deleted == 0,
StorageObjects.relative_path.like(f"{child_prefix}%"),
~StorageObjects.relative_path.like(f"{child_prefix}%/%"),
StorageObjects.usage_type.notin_(TREE_EXCLUDED_USAGE_TYPES),
).limit(1)
)
directory["has_children"] = has_children is not None
sorted_dirs = sorted(directories.values(), key=lambda item: item["path"])
return {
"request_id": context.request_id,
"data": {"directories": sorted_dirs},
@@ -873,11 +1009,12 @@ async def list_scripts(
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
statement = (
select(Scripts, StorageObjects)
select(Scripts, StorageObjects, Users.display_name)
.join(
StorageObjects,
StorageObjects.storage_object_id == Scripts.current_object_id,
)
.outerjoin(Users, Users.user_id == Scripts.owner_user_id)
.where(
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.status == "active",
@@ -888,7 +1025,8 @@ 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, owner_display_name=owner_display_name)
for script, storage_object, owner_display_name in rows
],
"meta": {"count": len(rows)},
}
@@ -971,24 +1109,16 @@ async def update_script(
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
# Look up the Scripts row on its own: jupyter-only scripts do not
# have a StorageObjects row to JOIN against, and update is an
# in-place overwrite of the same jupyter path, so we do not need
# any object-store metadata.
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()
# Load the working-copy StorageObject as well as the Scripts row.
# Jupyter-only scripts may not have a StorageObjects row; allow that
# case and fall back to the flat _jupyter_path() name.
script, storage_object = await get_script_row(
script_id,
context,
session,
for_update=True,
allow_missing_storage_object=True,
)
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,
@@ -997,21 +1127,23 @@ async def update_script(
content = validate_script_content(payload.content, script.script_type)
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
jupyter_name = _jupyter_path(script.script_type, script.script_id)
jupyter_path = _derive_jupyter_path(
storage_object, workspace_id, script.script_type, script.script_id
)
try:
if script.script_type == "notebook":
notebook = json.loads(content.decode("utf-8"))
jupyter_resp = await runtime_client.create_notebook(
workspace_id,
name=jupyter_name,
name=jupyter_path,
cells=notebook.get("cells"),
)
else:
jupyter_resp = await runtime_client.upload_file(
workspace_id,
name=jupyter_name,
name=jupyter_path,
content=content.decode("utf-8"),
content_type=(mimetypes.guess_type(jupyter_name)[0] or "text/plain"),
content_type=(mimetypes.guess_type(jupyter_path)[0] or "text/plain"),
)
except RuntimeClientError as exc:
raise HTTPException(
@@ -1022,8 +1154,8 @@ async def update_script(
script.updated_at = datetime.now(UTC).replace(tzinfo=None)
storage_data = {
"storage_object_id": script.current_object_id,
"relative_path": jupyter_name,
"object_key": f"{workspace_id}/{jupyter_name}",
"relative_path": jupyter_path,
"object_key": f"{workspace_id}/{jupyter_path}",
"content_hash": hashlib.sha256(content).hexdigest(),
"size_bytes": len(content),
}
@@ -1083,23 +1215,16 @@ async def delete_script(
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
# Scripts created via the jupyter path do not have a corresponding
# StorageObjects row, so we look up the Scripts row on its own and
# forward the delete to the workspace's live Jupyter instance.
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()
# Load the working-copy StorageObject if it exists. Jupyter-only
# scripts may not have a StorageObjects row; fall back to the flat
# _jupyter_path() name in that case.
script, storage_object = await get_script_row(
script_id,
context,
session,
for_update=True,
allow_missing_storage_object=True,
)
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,
@@ -1107,17 +1232,23 @@ async def delete_script(
)
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
jupyter_path = _derive_jupyter_path(
storage_object, workspace_id, script.script_type, script.script_id
)
try:
await runtime_client.delete_file(
context.workspace.workspace_id,
name=_jupyter_path(script.script_type, script.script_id),
)
await runtime_client.delete_file(workspace_id, name=jupyter_path)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
detail=exc.detail,
) from exc
if storage_object:
storage_object.object_status = "deleted"
storage_object.is_deleted = 1
storage_object.deleted_at = datetime.now(UTC).replace(tzinfo=None)
script.status = "deleted"
script.deleted_at = datetime.now(UTC).replace(tzinfo=None)
return {
@@ -1142,25 +1273,16 @@ async def publish_version(
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
# Jupyter-only scripts do not have a StorageObjects row, so the
# legacy get_script_row helper raises 409 before we even get here.
# Look up the Scripts row on its own — version publication is a
# metadata operation, we do not need the working-copy object
# metadata.
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()
# Load the working-copy StorageObject if it exists. Jupyter-only
# scripts may not have a StorageObjects row; fall back to the flat
# _jupyter_path() name when reading from Jupyter.
script, storage_object = await get_script_row(
script_id,
context,
session,
for_update=True,
allow_missing_storage_object=True,
)
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,
@@ -1180,9 +1302,11 @@ async def publish_version(
# files come back as a UTF-8 string.
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
jupyter_name = _jupyter_path(script.script_type, script.script_id)
jupyter_path = _derive_jupyter_path(
storage_object, workspace_id, 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_path)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
@@ -1241,7 +1365,7 @@ async def publish_version(
artifact_object_id=artifact_data["storage_object_id"],
version_no=version_no,
version_label=f"v{version_no}.0",
source_path=jupyter_name,
source_path=jupyter_path,
artifact_path=artifact_data["storage_uri"],
content_hash=content_hash,
file_size_bytes=len(content),
backend/src/backend/services/storage.py
+3 -2
View File
@@ -47,6 +47,7 @@ from common.storage.schemas import (
DownloadUrlRequest,
ServerObjectRequest,
)
from common.storage import build_storage_uri
# ── shared low-level helpers (module-private) ────────────────────────────
@@ -90,7 +91,7 @@ def _build_storage_object(
bucket_name=upload.bucket_name,
object_key=upload.object_key,
object_key_hash=upload.object_key_hash,
storage_uri=f"s3://{upload.bucket_name}/{upload.object_key}",
storage_uri=build_storage_uri(upload.bucket_name, upload.object_key),
file_name=safe_name,
file_extension=PurePosixPath(safe_name).suffix.lower() or None,
mime_type=content_type,
@@ -519,4 +520,4 @@ async def soft_delete_object(
"trash_key": item.trash_key,
"trash_bucket": settings.s3_trash_bucket,
}
}
}
backend/src/backend/storage_api.py
+2 -2
View File
@@ -19,7 +19,7 @@ from common.db.models import (
Workspaces,
)
from common.ids import new_ulid
from common.storage import actual_bucket_name
from common.storage import actual_bucket_name, build_storage_uri
from common.storage.schemas import (
CreateUploadRequest,
DownloadUrlRequest,
@@ -350,7 +350,7 @@ async def upload_bytes_to_session(
bucket_name=upload.bucket_name,
object_key=upload.object_key,
object_key_hash=upload.object_key_hash,
storage_uri=f"s3://{upload.bucket_name}/{upload.object_key}",
storage_uri=build_storage_uri(upload.bucket_name, upload.object_key),
file_name=file_name,
file_extension=PurePosixPath(file_name).suffix.lower() or None,
mime_type=upload.content_type,