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

refactor: directory

Browse Source
This commit is contained in:
tao.chen
2026-08-12 12:23:19 +08:00
parent a378ea5352
commit 3981c32781
4 changed files with 265 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/backend/runtime_client.py
+59
View File
@@ -288,5 +288,64 @@ class RuntimeClient:
ws = await self._ensure_workspace(workspace_id)
return await self._jupyter_request(workspace_id, ws, "GET", name)
async def create_directory(
self,
workspace_id: str,
*,
name: str,
) -> dict[str, Any]:
"""Create a directory in the workspace Jupyter.
Uses ``PUT /api/contents/{name}`` with body ``{"type": "directory"}``.
The directory's on-disk name in Jupyter is ``name``; callers that
want the display basename preserved separately should pass a ULID
(or other stable identifier) as ``name`` and keep the human-readable
basename in the database.
"""
ws = await self._ensure_workspace(workspace_id)
return await self._jupyter_request(
workspace_id, ws, "PUT", name, body={"type": "directory"}
)
async def delete_directory(
self,
workspace_id: str,
*,
name: str,
) -> None:
"""Delete a directory from the workspace Jupyter.
Uses ``DELETE /api/contents/{name}`` on the workspace Jupyter.
Jupyter returns ``204 No Content`` on success; any 4xx/5xx is
surfaced as :class:`RuntimeClientError`. Non-recursive — Jupyter
returns ``409 Conflict`` when the directory is not empty; the
caller is responsible for emptying the directory first.
"""
ws = await self._ensure_workspace(workspace_id)
await self._jupyter_request(workspace_id, ws, "DELETE", name)
async def ensure_directory(
self,
workspace_id: str,
name: str,
) -> None:
"""Ensure a directory exists in the workspace Jupyter.
Lazy-backfill primitive used during create flows: performs a
``GET /api/contents/{name}`` first, returning immediately if the
directory already exists. On ``404 Not Found`` it falls through
to :meth:`create_directory`. Any other error is propagated as a
:class:`RuntimeClientError`. The GET-first pattern avoids the
ambiguity of calling ``PUT /api/contents/{name}`` on a path that
may already exist.
"""
ws = await self._ensure_workspace(workspace_id)
try:
await self._jupyter_request(workspace_id, ws, "GET", name)
except RuntimeClientError as exc:
if exc.status_code == 404:
await self.create_directory(workspace_id, name=name)
return
raise
__all__ = ["RuntimeClient", "RuntimeClientError"]
backend/src/backend/scripts.py
+194 -61
View File
@@ -303,12 +303,28 @@ async def create_script_record(
context: RequestContext,
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)
)
child_path = f"{folder}/{name}" if folder else name
parent = normalize_user_path(parent_path) if parent_path else ""
scoped_prefix = user_relative_path(context)
if parent:
parent_dir_row = await session.scalar(
select(StorageObjects).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.object_type == "directory",
StorageObjects.relative_path == f"{scoped_prefix}/{parent}",
)
)
if parent_dir_row is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"parent directory not found",
)
parent_ulid = parent_dir_row.storage_object_id
else:
parent_ulid = None
display_parent = parent
child_path = f"{display_parent}/{name}" if display_parent else name
relative_path = user_relative_path(context, child_path)
logger.debug(relative_path)
@@ -319,6 +335,7 @@ async def create_script_record(
# claim the same display name within the same workspace.
script_id = new_ulid()
jupyter_name = _jupyter_path(script_type, script_id)
nested_jupyter_name = f"{parent_ulid}/{jupyter_name}" if parent_ulid else jupyter_name
name_clash = await session.scalar(
select(Scripts.script_id).where(
Scripts.workspace_id == context.workspace.workspace_id,
@@ -340,19 +357,28 @@ async def create_script_record(
workspace_id = context.workspace.workspace_id
content_hash = hashlib.sha256(content).hexdigest()
size_bytes = len(content)
if parent_ulid:
try:
await runtime_client.ensure_directory(workspace_id, parent_ulid)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
detail=exc.detail,
) from exc
try:
if script_type == "notebook":
notebook = json.loads(content.decode("utf-8"))
logger.debug(notebook)
jupyter_resp = await runtime_client.create_notebook(
workspace_id,
name=jupyter_name,
name=nested_jupyter_name,
cells=notebook.get("cells"),
)
else:
jupyter_resp = await runtime_client.upload_file(
workspace_id,
name=jupyter_name,
name=nested_jupyter_name,
content=content.decode("utf-8"),
content_type=(mimetypes.guess_type(jupyter_name)[0] or "text/plain"),
)
@@ -370,9 +396,11 @@ async def create_script_record(
# file is queryable as a workspace file from the user's POV; the
# storage_uri points at where the replicated bytes will land.
object_id = new_ulid()
object_key = f"{workspace_id}/{jupyter_name}"
object_key = f"{workspace_id}/{nested_jupyter_name}"
bucket_name = settings.s3_workspace_bucket
relative_path = user_relative_path(context, jupyter_name)
relative_path = user_relative_path(
context, f"{parent}/{jupyter_name}" if parent else jupyter_name
)
mime_type = mimetypes.guess_type(jupyter_name)[0]
storage_object = StorageObjects(
storage_object_id=object_id,
@@ -407,11 +435,26 @@ async def create_script_record(
visibility=visibility,
status="active",
)
session.add(storage_object)
session.add(script)
await session.flush()
await session.refresh(storage_object)
await session.refresh(script)
try:
session.add(storage_object)
session.add(script)
await session.flush()
await session.refresh(storage_object)
await session.refresh(script)
except Exception:
# Best-effort compensating cleanup: remove the Jupyter file we
# just created so a failed flush does not leave an orphan on disk.
try:
await runtime_client.delete_file(workspace_id, name=nested_jupyter_name)
except RuntimeClientError as cleanup_exc:
if cleanup_exc.status_code == 404:
pass
else:
logger.warning(
f"failed to clean up Jupyter file {nested_jupyter_name} "
f"after DB flush error: {cleanup_exc.status_code} {cleanup_exc.detail}"
)
raise
return script, storage_object
@@ -582,6 +625,7 @@ async def get_workspace_tree(
)
async def create_workspace_directory(
payload: CreateWorkspaceDirectoryRequest,
request: Request,
context: RequestContext = Depends(request_context),
session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
@@ -591,24 +635,27 @@ async def create_workspace_directory(
relative_path = user_relative_path(context, child_path)
scoped_prefix = user_relative_path(context)
path_hash = hashlib.sha256(relative_path.encode("utf-8")).digest()
# Validate parent exists: there must be at least one StorageObject whose
# 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(
# Only an available directory row at the exact parent path counts as
# a parent. Capture its ULID so the new Jupyter directory can be nested
# under it on disk.
parent_dir_row = await session.scalar(
select(StorageObjects).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
(StorageObjects.relative_path == parent_relative)
| StorageObjects.relative_path.like(f"{parent_relative}/%"),
StorageObjects.object_type == "directory",
StorageObjects.relative_path == parent_relative,
)
)
if existing_parent is None:
if parent_dir_row is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
"parent directory not found",
)
parent_ulid = parent_dir_row.storage_object_id
else:
parent_ulid = None
# Conflict check uses the unique index on (workspace_id, storage_backend, path_hash).
# A soft-deleted row at the same path can be revived; an available row is a conflict.
existing = await session.scalar(
@@ -626,11 +673,36 @@ async def create_workspace_directory(
"a file or directory with the same path already exists",
)
# Generate the Jupyter-facing directory ULID up front so the DB id and
# the on-disk directory name match. Create the directory in Jupyter
# before persisting the DB row; if persistence fails we clean up.
dir_id = new_ulid()
jupyter_path = f"{parent_ulid}/{dir_id}" if parent_ulid else dir_id
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
if parent_ulid:
try:
await runtime_client.ensure_directory(workspace_id, parent_ulid)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
detail=exc.detail,
) from exc
try:
await runtime_client.create_directory(workspace_id, name=jupyter_path)
except RuntimeClientError as exc:
raise HTTPException(
status_code=exc.status_code,
detail=exc.detail,
) from exc
if existing is None:
directory = StorageObjects(
storage_object_id=new_ulid(),
storage_object_id=dir_id,
workspace_id=context.workspace.workspace_id,
storage_backend="rustfs",
storage_backend=settings.storage_backend,
object_type="directory",
usage_type="working_copy",
storage_uri=f"inline://directory/{relative_path}",
@@ -644,7 +716,6 @@ async def create_workspace_directory(
)
session.add(directory)
else:
# Revive the soft-deleted row, keeping its original storage_object_id.
directory = existing
directory.object_status = "available"
directory.is_deleted = 0
@@ -657,7 +728,22 @@ async def create_workspace_directory(
directory.visibility = "private"
directory.created_by = context.user.user_id
await session.flush()
try:
await session.flush()
except Exception:
# Best-effort compensating cleanup: remove the Jupyter directory we
# just created so a failed flush does not leave an orphan on disk.
try:
await runtime_client.delete_directory(workspace_id, name=jupyter_path)
except RuntimeClientError as cleanup_exc:
if cleanup_exc.status_code == 404:
pass
else:
logger.warning(
f"failed to clean up Jupyter directory {jupyter_path} "
f"after DB flush error: {cleanup_exc.status_code} {cleanup_exc.detail}"
)
raise
return {
"request_id": context.request_id,
"data": {
@@ -679,55 +765,102 @@ async def delete_workspace_directory(
) -> dict[str, Any]:
directory_path = normalize_user_path(path, allow_empty=False)
# The original query JOINed StorageObjects to filter scripts by
# ``relative_path`` prefix. Jupyter-only scripts have no
# StorageObject row, and the Scripts schema does not track
# directory membership, so we cannot honour the directory filter
# for jupyter-created files. We therefore enumerate all active
# scripts the user owns in the workspace and forward each delete
# to the runtime. ``directory_path`` is still returned in the
# response for API compatibility.
rows = (
scoped_prefix = user_relative_path(context)
target_relative = f"{scoped_prefix}/{directory_path}"
target_dir_row = await session.scalar(
select(StorageObjects).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_type == "directory",
StorageObjects.relative_path == target_relative,
StorageObjects.object_status == "available",
)
)
if target_dir_row is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "directory not found")
target_ulid = target_dir_row.storage_object_id
child_prefix = f"{target_relative}/"
descendants = (
(
await session.execute(
select(Scripts).where(
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.owner_user_id == context.user.user_id,
Scripts.status == "active",
)
select(StorageObjects).where(
StorageObjects.workspace_id == context.workspace.workspace_id,
StorageObjects.object_status == "available",
StorageObjects.relative_path.like(f"{child_prefix}%"),
).order_by(func.length(StorageObjects.relative_path).desc())
)
)
.scalars()
.all()
)
runtime_client = request.app.state.runtime_client
workspace_id = context.workspace.workspace_id
for script in rows:
# Best-effort: try to delete from jupyter, but do not let one
# failure abort the rest. The jupyter call will 404 if the
# file is not actually in the workspace (e.g. legacy scripts
# whose on-disk filename we do not know); we treat that as a
# no-op and still mark the row deleted.
try:
await runtime_client.delete_file(
workspace_id,
name=_jupyter_path(script.script_type, script.script_id),
)
except RuntimeClientError as exc:
if exc.status_code != 404:
logger.warning(
f"delete_workspace_directory: jupyter delete "
f"failed for {script.script_id} ({script.script_name}): "
f"{exc.status_code} {exc.detail}"
deleted_scripts = 0
for descendant in descendants:
if descendant.object_type == "file":
jupyter_path = descendant.object_key.removeprefix(f"{workspace_id}/")
try:
await runtime_client.delete_file(workspace_id, name=jupyter_path)
except RuntimeClientError as exc:
if exc.status_code != 404:
logger.warning(
f"delete_workspace_directory: jupyter delete failed "
f"for file {descendant.storage_object_id}: "
f"{exc.status_code} {exc.detail}"
)
script = await session.scalar(
select(Scripts).where(
Scripts.workspace_id == context.workspace.workspace_id,
Scripts.current_object_id == descendant.storage_object_id,
)
script.status = "deleted"
script.deleted_at = datetime.now(UTC).replace(tzinfo=None)
)
if script is not None:
script.status = "deleted"
script.deleted_at = datetime.now(UTC).replace(tzinfo=None)
deleted_scripts += 1
descendant.object_status = "deleted"
descendant.is_deleted = 1
descendant.deleted_at = datetime.now(UTC).replace(tzinfo=None)
else:
try:
await runtime_client.delete_directory(
workspace_id, name=descendant.storage_object_id
)
except RuntimeClientError as exc:
if exc.status_code != 404:
logger.warning(
f"delete_workspace_directory: jupyter delete failed "
f"for directory {descendant.storage_object_id}: "
f"{exc.status_code} {exc.detail}"
)
descendant.object_status = "deleted"
descendant.is_deleted = 1
descendant.deleted_at = datetime.now(UTC).replace(tzinfo=None)
try:
await runtime_client.delete_directory(workspace_id, name=target_ulid)
except RuntimeClientError as exc:
if exc.status_code != 404:
logger.warning(
f"delete_workspace_directory: jupyter delete failed "
f"for target directory {target_ulid}: "
f"{exc.status_code} {exc.detail}"
)
target_dir_row.object_status = "deleted"
target_dir_row.is_deleted = 1
target_dir_row.deleted_at = datetime.now(UTC).replace(tzinfo=None)
await session.flush()
return {
"request_id": context.request_id,
"data": {
"path": directory_path,
"status": "deleted",
"deleted_scripts": len(rows),
"deleted_scripts": deleted_scripts,
"versions_preserved": True,
},
"meta": {},
frontend/app/components/platform/ScriptExplorer.tsx
+5 -1
View File
@@ -121,10 +121,14 @@ export function ScriptExplorer({
role_code: null,
is_system_admin: false,
} as AuthUser);
const inferred = inferredDirectories(groupScripts);
groups.push({
user: groupUser,
scripts: groupScripts,
directories: inferredDirectories(groupScripts),
directories:
ownerUserId === user?.user_id
? mergeDirectories(directories, inferred)
: inferred,
});
}
frontend/app/features/platform/ScriptsPage.tsx
+7 -2
View File
@@ -189,6 +189,11 @@ export default function ScriptsPage() {
void submitPublish(publish.releaseNote, publish.visibility);
};
const triggerUpload = (parentPath?: string) => {
chooseUpload(parentPath);
uploadInputRef.current?.click();
};
return (
<section className="workspace-layout">
<ScriptExplorer
@@ -203,11 +208,11 @@ export default function ScriptsPage() {
keyword={keyword}
onKeywordChange={setKeyword}
onRefresh={() => void load(true)}
onUpload={() => chooseUpload("")}
onUpload={() => triggerUpload("")}
onOpenCreateDialog={(parentPath, scriptType) =>
openCreateDialog(parentPath, scriptType)}
onOpenFolderDialog={(parentPath) => openFolderDialog(parentPath)}
onChooseUpload={(parentPath) => chooseUpload(parentPath)}
onChooseUpload={(parentPath) => triggerUpload(parentPath)}
onContextMenu={showContextMenu}
onSelect={openTab}
uploadInputRef={uploadInputRef}