171 lines
6.3 KiB
Python
171 lines
6.3 KiB
Python
# #region storage_routes [C:5] [TYPE Module] [SEMANTICS fastapi, storage, api, upload, download]
|
|
#
|
|
# @BRIEF API endpoints for file storage management (backups and repositories).
|
|
# @LAYER API
|
|
# @RELATION DEPENDS_ON -> [StorageModels]
|
|
# @RELATION DEPENDS_ON -> [StoragePlugin]
|
|
#
|
|
# @INVARIANT All paths must be validated against path traversal.
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
|
|
from ...core.logger import belief_scope
|
|
from ...dependencies import get_plugin_loader, has_permission
|
|
from ...models.storage import FileCategory, StoredFile
|
|
from ...plugins.storage.plugin import StoragePlugin
|
|
|
|
router = APIRouter(tags=["storage"])
|
|
|
|
# #region list_files [C:3] [TYPE Function]
|
|
# @BRIEF List all files and directories in the storage system.
|
|
#
|
|
# @PRE None.
|
|
# @POST Returns a list of StoredFile objects.
|
|
#
|
|
# @RELATION DEPENDS_ON -> [StoragePlugin]
|
|
@router.get("/files", response_model=list[StoredFile])
|
|
async def list_files(
|
|
category: FileCategory | None = None,
|
|
path: str | None = None,
|
|
recursive: bool = False,
|
|
plugin_loader=Depends(get_plugin_loader),
|
|
_ = Depends(has_permission("plugin:storage", "READ"))
|
|
):
|
|
with belief_scope("list_files"):
|
|
storage_plugin: StoragePlugin = plugin_loader.get_plugin("storage-manager")
|
|
if not storage_plugin:
|
|
raise HTTPException(status_code=500, detail="Storage plugin not loaded")
|
|
return storage_plugin.list_files(category, path, recursive)
|
|
# #endregion list_files
|
|
|
|
# #region upload_file [C:3] [TYPE Function]
|
|
# @BRIEF Upload a file to the storage system.
|
|
#
|
|
# @PRE category must be a valid FileCategory.
|
|
# @PRE file must be a valid UploadFile.
|
|
# @POST Returns the StoredFile object of the uploaded file.
|
|
#
|
|
#
|
|
# @SIDE_EFFECT Writes file to the filesystem.
|
|
#
|
|
# @RELATION DEPENDS_ON -> [StoragePlugin]
|
|
@router.post("/upload", response_model=StoredFile, status_code=201)
|
|
async def upload_file(
|
|
category: FileCategory = Form(...),
|
|
path: str | None = Form(None),
|
|
file: UploadFile = File(...),
|
|
plugin_loader=Depends(get_plugin_loader),
|
|
_ = Depends(has_permission("plugin:storage", "WRITE"))
|
|
):
|
|
with belief_scope("upload_file"):
|
|
storage_plugin: StoragePlugin = plugin_loader.get_plugin("storage-manager")
|
|
if not storage_plugin:
|
|
raise HTTPException(status_code=500, detail="Storage plugin not loaded")
|
|
try:
|
|
return await storage_plugin.save_file(file, category, path)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
# #endregion upload_file
|
|
|
|
# #region delete_file [C:3] [TYPE Function]
|
|
# @BRIEF Delete a specific file or directory.
|
|
#
|
|
# @PRE category must be a valid FileCategory.
|
|
# @POST Item is removed from storage.
|
|
#
|
|
#
|
|
# @SIDE_EFFECT Deletes item from the filesystem.
|
|
#
|
|
# @RELATION DEPENDS_ON -> [StoragePlugin]
|
|
@router.delete("/files/{category}/{path:path}", status_code=204)
|
|
async def delete_file(
|
|
category: FileCategory,
|
|
path: str,
|
|
plugin_loader=Depends(get_plugin_loader),
|
|
_ = Depends(has_permission("plugin:storage", "WRITE"))
|
|
):
|
|
with belief_scope("delete_file"):
|
|
storage_plugin: StoragePlugin = plugin_loader.get_plugin("storage-manager")
|
|
if not storage_plugin:
|
|
raise HTTPException(status_code=500, detail="Storage plugin not loaded")
|
|
try:
|
|
storage_plugin.delete_file(category, path)
|
|
except FileNotFoundError:
|
|
raise HTTPException(status_code=404, detail="File not found")
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
# #endregion delete_file
|
|
|
|
# #region download_file [C:3] [TYPE Function]
|
|
# @BRIEF Retrieve a file for download.
|
|
#
|
|
# @PRE category must be a valid FileCategory.
|
|
# @POST Returns a FileResponse.
|
|
#
|
|
#
|
|
# @RELATION DEPENDS_ON -> [StoragePlugin]
|
|
@router.get("/download/{category}/{path:path}")
|
|
async def download_file(
|
|
category: FileCategory,
|
|
path: str,
|
|
plugin_loader=Depends(get_plugin_loader),
|
|
_ = Depends(has_permission("plugin:storage", "READ"))
|
|
):
|
|
with belief_scope("download_file"):
|
|
storage_plugin: StoragePlugin = plugin_loader.get_plugin("storage-manager")
|
|
if not storage_plugin:
|
|
raise HTTPException(status_code=500, detail="Storage plugin not loaded")
|
|
try:
|
|
abs_path = storage_plugin.get_file_path(category, path)
|
|
filename = Path(path).name
|
|
return FileResponse(path=abs_path, filename=filename)
|
|
except FileNotFoundError:
|
|
raise HTTPException(status_code=404, detail="File not found")
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
# #endregion download_file
|
|
|
|
# #region get_file_by_path [C:3] [TYPE Function]
|
|
# @BRIEF Retrieve a file by validated absolute/relative path under storage root.
|
|
#
|
|
# @PRE path must resolve under configured storage root.
|
|
# @POST Returns a FileResponse for existing files.
|
|
#
|
|
#
|
|
# @RELATION DEPENDS_ON -> [StoragePlugin]
|
|
@router.get("/file")
|
|
async def get_file_by_path(
|
|
path: str,
|
|
plugin_loader=Depends(get_plugin_loader),
|
|
_ = Depends(has_permission("plugin:storage", "READ"))
|
|
):
|
|
with belief_scope("get_file_by_path"):
|
|
storage_plugin: StoragePlugin = plugin_loader.get_plugin("storage-manager")
|
|
if not storage_plugin:
|
|
raise HTTPException(status_code=500, detail="Storage plugin not loaded")
|
|
|
|
requested_path = (path or "").strip()
|
|
if not requested_path:
|
|
raise HTTPException(status_code=400, detail="Path is required")
|
|
|
|
try:
|
|
candidate = Path(requested_path)
|
|
if candidate.is_absolute():
|
|
abs_path = storage_plugin.validate_path(candidate)
|
|
else:
|
|
storage_root = storage_plugin.get_storage_root()
|
|
abs_path = storage_plugin.validate_path(storage_root / candidate)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
if not abs_path.exists() or not abs_path.is_file():
|
|
raise HTTPException(status_code=404, detail="File not found")
|
|
|
|
return FileResponse(path=str(abs_path), filename=abs_path.name)
|
|
# #endregion get_file_by_path
|
|
|
|
# #endregion storage_routes
|