feat: Node file streaming, recursive listing, file type detection, and file delete

- Stream file GET/POST through controller without buffering in memory
- Add recursive and subdirectory filtering to node file listing
- Replace file extension with magic-based file type detection
- Add DELETE endpoint for node and project files
- Include directories in listing response
- Add params and stream support to http_query
- Fix lambda closures, streamer exception scope, and delete error codes
This commit is contained in:
YueGuobin committed 2026-06-09 22:52:50 +08:00
1 parent c70a4660c1
commit cbb21e8e40
8 files changed
+289 -136

No files matched your search

+35 -56
View File
@@ -19,13 +19,14 @@ API routes for projects.
"""
import os
import shutil
import urllib.parse
import logging
log = logging.getLogger()
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status, Query
from fastapi.encoders import jsonable_encoder
from fastapi.responses import FileResponse
from typing import List
@@ -129,59 +130,6 @@ async def delete_compute_project(project: Project = Depends(dep_project)) -> Non
await project.delete()
ProjectManager.instance().remove_project(project.id)
# @Route.get(
# r"/projects/{project_id}/notifications",
# description="Receive notifications about the project",
# parameters={
# "project_id": "Project UUID",
# },
# status_codes={
# 200: "End of stream",
# 404: "The project doesn't exist"
# })
# async def notification(request, response):
#
# pm = ProjectManager.instance()
# project = pm.get_project(request.match_info["project_id"])
#
# response.content_type = "application/json"
# response.set_status(200)
# response.enable_chunked_encoding()
#
# response.start(request)
# queue = project.get_listen_queue()
# ProjectHandler._notifications_listening.setdefault(project.id, 0)
# ProjectHandler._notifications_listening[project.id] += 1
# await response.write("{}\n".format(json.dumps(ProjectHandler._getPingMessage())).encode("utf-8"))
# while True:
# try:
# (action, msg) = await asyncio.wait_for(queue.get(), 5)
# if hasattr(msg, "asdict"):
# msg = json.dumps({"action": action, "event": msg.asdict()}, sort_keys=True)
# else:
# msg = json.dumps({"action": action, "event": msg}, sort_keys=True)
# log.debug("Send notification: %s", msg)
# await response.write(("{}\n".format(msg)).encode("utf-8"))
# except asyncio.TimeoutError:
# await response.write("{}\n".format(json.dumps(ProjectHandler._getPingMessage())).encode("utf-8"))
# project.stop_listen_queue(queue)
# if project.id in ProjectHandler._notifications_listening:
# ProjectHandler._notifications_listening[project.id] -= 1
# def _getPingMessage(cls):
# """
# Ping messages are regularly sent to the client to
# keep the connection open. We send with it some information about server load.
#
# :returns: hash
# """
# stats = {}
# # Non blocking call in order to get cpu usage. First call will return 0
# stats["cpu_usage_percent"] = CpuPercent.get(interval=None)
# stats["memory_usage_percent"] = psutil.virtual_memory().percent
# stats["disk_usage_percent"] = psutil.disk_usage(get_default_project_directory()).percent
# return {"action": "ping", "event": stats}
@router.get("/projects/{project_id}/files", response_model=List[schemas.ProjectFile])
async def get_compute_project_files(project: Project = Depends(dep_project)) -> List[schemas.ProjectFile]:
@@ -196,14 +144,16 @@ async def get_compute_project_files(project: Project = Depends(dep_project)) ->
async def get_compute_node_files(
node_type: str,
node_id: str,
project: Project = Depends(dep_project)
project: Project = Depends(dep_project),
path: str = Query("", description="Subdirectory path within node directory"),
recursive: bool = Query(False, description="Recursively list all files")
) -> List[schemas.NodeFile]:
"""
Return files belonging to a specific node with detailed metadata.
"""
node_path = f"project-files/{node_type}/{node_id}"
return await project.list_node_files(node_path)
return await project.list_node_files(node_path, subpath=path, recursive=recursive)
@router.get("/projects/{project_id}/files/{file_path:path}")
@@ -255,3 +205,32 @@ async def write_compute_project_file(
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
except PermissionError:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
@router.delete("/projects/{project_id}/files/{file_path:path}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_compute_project_file(
file_path: str,
project: Project = Depends(dep_project)
) -> None:
file_path = urllib.parse.unquote(file_path)
path = os.path.normpath(file_path)
if not is_safe_path(path, project.path):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
path = os.path.join(project.path, path)
if not os.path.exists(path):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
try:
if os.path.isdir(path):
shutil.rmtree(path)
else:
os.remove(path)
except FileNotFoundError:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
except PermissionError:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
except OSError as e:
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
+70 -10
View File
@@ -24,6 +24,7 @@ import ipaddress
from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect, Request, Response, status, Query, HTTPException
from fastapi.encoders import jsonable_encoder
from fastapi.responses import StreamingResponse
from fastapi.routing import APIRoute
from typing import List, Callable, Optional
from uuid import UUID
@@ -523,17 +524,30 @@ async def delete_disk_image(
@router.get("/{node_id}/files", response_model=List[schemas.NodeFile], dependencies=[Depends(has_privilege("Node.Audit"))])
async def list_node_files(node: Node = Depends(dep_node)) -> List[schemas.NodeFile]:
async def list_node_files(
node: Node = Depends(dep_node),
path: str = Query("", description="Subdirectory path within node directory"),
recursive: bool = Query(False, description="Recursively list all files")
) -> List[schemas.NodeFile]:
"""
List files in a node directory with detailed metadata.
By default lists only the current directory level (non-recursive).
Use recursive=true for a full recursive listing.
Required privilege: Node.Audit
"""
node_type = node.node_type
url = f"/projects/{node.project.id}/nodes/{node_type}/{node.id}/files"
params = {}
if path:
params["path"] = path
if recursive:
params["recursive"] = "true"
res = await node.compute.http_query(
"GET",
f"/projects/{node.project.id}/nodes/{node_type}/{node.id}/files",
"GET", url,
params=params if params else None,
timeout=None
)
return res.json
@@ -550,14 +564,32 @@ async def get_file(file_path: str, node: Node = Depends(dep_node)) -> Response:
path = force_unix_path(file_path)
# Raise error if user try to escape
if path[0] == ".":
if path.startswith(".."):
raise ControllerForbiddenError("It is forbidden to get a file outside the project directory")
node_type = node.node_type
path = f"/project-files/{node_type}/{node.id}/{path}"
res = await node.compute.http_query("GET", f"/projects/{node.project.id}/files{path}", timeout=None, raw=True)
return Response(res.body, media_type="application/octet-stream", status_code=res.status)
compute_resp = await node.compute.http_query(
"GET", f"/projects/{node.project.id}/files{path}",
timeout=None, stream=True
)
async def streamer():
try:
async for chunk in compute_resp.content.iter_chunked(65536):
yield chunk
except (IOError, OSError, asyncio.TimeoutError) as e:
log.error(f"Error streaming file '{path}' from compute: {e}")
raise
finally:
compute_resp.close()
return StreamingResponse(
streamer(),
media_type="application/octet-stream",
status_code=compute_resp.status,
)
@router.post(
@@ -575,15 +607,43 @@ async def post_file(file_path: str, request: Request, node: Node = Depends(dep_n
path = force_unix_path(file_path)
# Raise error if user try to escape
if path[0] == ".":
if path.startswith(".."):
raise ControllerForbiddenError("Cannot write outside the node directory")
node_type = node.node_type
path = f"/project-files/{node_type}/{node.id}/{path}"
data = await request.body() # FIXME: are we handling timeout or large files correctly?
await node.compute.http_query("POST", f"/projects/{node.project.id}/files{path}", data=data, timeout=None, raw=True)
# FIXME: response with correct status code (from compute)
# Stream request body directly to compute node
await node.compute.http_query(
"POST", f"/projects/{node.project.id}/files{path}",
data=request.stream(), timeout=None
)
@router.delete(
"/{node_id}/files/{file_path:path}",
status_code=status.HTTP_204_NO_CONTENT,
dependencies=[Depends(has_privilege("Node.Modify"))]
)
async def delete_node_file(file_path: str, node: Node = Depends(dep_node)) -> None:
"""
Delete a file from the node directory.
Required privilege: Node.Modify
"""
path = force_unix_path(file_path)
if path.startswith(".."):
raise ControllerForbiddenError("It is forbidden to delete a file outside the project directory")
node_type = node.node_type
path = f"/project-files/{node_type}/{node.id}/{path}"
await node.compute.http_query(
"DELETE", f"/projects/{node.project.id}/files{path}",
timeout=None
)
@router.websocket("/{node_id}/console/ws")