Add browser Python editor with Pyodide, user auth, and workspace API
- FastAPI serves static UI, file CRUD under code/ and read-only lib/ - Pyodide worker runs Python and Jedi completions in the browser - SQLite accounts: login/register, session cookies, superuser user management - Optional EDITOR_API_KEY, AUTH_* env vars, .env.example - Pipenv, pytest, Selenium smoke test, README Made-with: Cursor
This commit is contained in:
202
src/editor_app/services/filesystem.py
Normal file
202
src/editor_app/services/filesystem.py
Normal file
@@ -0,0 +1,202 @@
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
from editor_app import config
|
||||
from editor_app.models import FileInfo
|
||||
|
||||
LIB_DIR_NAME = "lib"
|
||||
WRITABLE_ROOTS = {"code"}
|
||||
|
||||
|
||||
def normalize_relative_path(relative_path: str) -> str:
|
||||
cleaned = (relative_path or "").strip().lstrip("/")
|
||||
if not cleaned:
|
||||
return ""
|
||||
|
||||
parts = [segment for segment in cleaned.split("/") if segment]
|
||||
if len(parts) >= 2 and parts[0] == "code":
|
||||
while len(parts) >= 2 and parts[0] == parts[1] == "code":
|
||||
parts.pop(1)
|
||||
return "/".join(parts)
|
||||
|
||||
|
||||
def resolve_workspace_path(relative_path: str) -> Path:
|
||||
relative_path = normalize_relative_path(relative_path)
|
||||
target_path = (config.WORKSPACE_ROOT / relative_path).resolve()
|
||||
try:
|
||||
target_path.relative_to(config.WORKSPACE_ROOT.resolve())
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail="Path escapes workspace") from exc
|
||||
return target_path
|
||||
|
||||
|
||||
def _is_path_in_lib(target_path: Path) -> bool:
|
||||
workspace = config.WORKSPACE_ROOT.resolve()
|
||||
lib_root = (workspace / LIB_DIR_NAME).resolve()
|
||||
try:
|
||||
target_path.resolve().relative_to(lib_root)
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def _ensure_not_lib_path(target_path: Path) -> None:
|
||||
if _is_path_in_lib(target_path):
|
||||
raise HTTPException(status_code=403, detail="lib is read-only")
|
||||
|
||||
|
||||
def _is_writable_path(target_path: Path) -> bool:
|
||||
workspace = config.WORKSPACE_ROOT.resolve()
|
||||
resolved = target_path.resolve()
|
||||
try:
|
||||
relative = resolved.relative_to(workspace)
|
||||
except ValueError:
|
||||
return False
|
||||
if not relative.parts:
|
||||
return False
|
||||
return relative.parts[0] in WRITABLE_ROOTS
|
||||
|
||||
|
||||
def _ensure_writable_path(target_path: Path) -> None:
|
||||
if not _is_writable_path(target_path):
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="Only code/ is writable (lib is read-only)",
|
||||
)
|
||||
|
||||
|
||||
def list_files(path: str = "") -> list[FileInfo]:
|
||||
path = normalize_relative_path(path)
|
||||
target_path = config.WORKSPACE_ROOT / path if path else config.WORKSPACE_ROOT
|
||||
if not target_path.exists() or not target_path.is_dir():
|
||||
raise HTTPException(status_code=404, detail="Directory not found")
|
||||
|
||||
files = []
|
||||
for item in sorted(target_path.iterdir()):
|
||||
if item.name.startswith("."):
|
||||
continue
|
||||
files.append(
|
||||
FileInfo(
|
||||
name=item.name,
|
||||
is_directory=item.is_dir(),
|
||||
size=item.stat().st_size if item.is_file() else None,
|
||||
)
|
||||
)
|
||||
return files
|
||||
|
||||
|
||||
def read_text_file(file_path: str) -> tuple[str, str]:
|
||||
target_path = resolve_workspace_path(file_path)
|
||||
if not target_path.exists():
|
||||
raise HTTPException(status_code=404, detail="File not found")
|
||||
if target_path.is_dir():
|
||||
raise HTTPException(status_code=400, detail="Path is a directory")
|
||||
try:
|
||||
content = target_path.read_text(encoding="utf-8")
|
||||
except UnicodeDecodeError as exc:
|
||||
raise HTTPException(status_code=400, detail="File is not a text file") from exc
|
||||
return content, target_path.name
|
||||
|
||||
|
||||
def save_text_file(file_path: str, content: str) -> str:
|
||||
target_path = resolve_workspace_path(file_path)
|
||||
_ensure_not_lib_path(target_path)
|
||||
_ensure_writable_path(target_path)
|
||||
target_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
target_path.write_text(content, encoding="utf-8")
|
||||
return target_path.name
|
||||
|
||||
|
||||
def delete_file(file_path: str) -> None:
|
||||
target_path = resolve_workspace_path(file_path)
|
||||
_ensure_not_lib_path(target_path)
|
||||
_ensure_writable_path(target_path)
|
||||
if not target_path.exists():
|
||||
raise HTTPException(status_code=404, detail="File not found")
|
||||
if target_path.is_dir():
|
||||
raise HTTPException(status_code=400, detail="Cannot delete directories")
|
||||
target_path.unlink()
|
||||
|
||||
|
||||
def move_path(source_path: str, destination_folder: str) -> tuple[str, str]:
|
||||
source = resolve_workspace_path(source_path)
|
||||
_ensure_not_lib_path(source)
|
||||
_ensure_writable_path(source)
|
||||
if not source.exists():
|
||||
raise HTTPException(status_code=404, detail="Source path not found")
|
||||
|
||||
destination_dir = (
|
||||
resolve_workspace_path(destination_folder)
|
||||
if destination_folder
|
||||
else config.WORKSPACE_ROOT
|
||||
)
|
||||
_ensure_not_lib_path(destination_dir)
|
||||
_ensure_writable_path(destination_dir)
|
||||
if not destination_dir.exists() or not destination_dir.is_dir():
|
||||
raise HTTPException(status_code=404, detail="Destination folder not found")
|
||||
|
||||
destination = destination_dir / source.name
|
||||
source_resolved = source.resolve()
|
||||
destination_resolved = destination.resolve()
|
||||
if destination_resolved == source_resolved:
|
||||
raise HTTPException(status_code=400, detail="Path is already in that folder")
|
||||
if source.is_dir():
|
||||
source_prefix = str(source_resolved) + "/"
|
||||
if str(destination_dir.resolve()).startswith(source_prefix):
|
||||
raise HTTPException(
|
||||
status_code=400, detail="Cannot move a folder into itself or its child"
|
||||
)
|
||||
if destination.exists():
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail="A path with that name already exists in destination",
|
||||
)
|
||||
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
source.rename(destination)
|
||||
moved_type = "folder" if destination.is_dir() else "file"
|
||||
return str(destination.relative_to(config.WORKSPACE_ROOT)), moved_type
|
||||
|
||||
|
||||
def create_folder(folder_path: str) -> str:
|
||||
target_path = resolve_workspace_path(folder_path)
|
||||
_ensure_not_lib_path(target_path)
|
||||
_ensure_writable_path(target_path)
|
||||
if target_path.exists():
|
||||
raise HTTPException(status_code=400, detail="Folder already exists")
|
||||
target_path.mkdir(parents=True, exist_ok=False)
|
||||
return target_path.name
|
||||
|
||||
|
||||
def delete_folder(folder_path: str) -> None:
|
||||
target_path = resolve_workspace_path(folder_path)
|
||||
_ensure_not_lib_path(target_path)
|
||||
_ensure_writable_path(target_path)
|
||||
if not target_path.exists():
|
||||
raise HTTPException(status_code=404, detail="Folder not found")
|
||||
if not target_path.is_dir():
|
||||
raise HTTPException(status_code=400, detail="Path is not a directory")
|
||||
shutil.rmtree(target_path)
|
||||
|
||||
|
||||
def collect_python_sources() -> dict[str, str]:
|
||||
"""Return all UTF-8 .py files under the workspace for browser-side Pyodide sync."""
|
||||
result: dict[str, str] = {}
|
||||
workspace = config.WORKSPACE_ROOT.resolve()
|
||||
if not workspace.exists():
|
||||
return result
|
||||
for path in workspace.rglob("*.py"):
|
||||
try:
|
||||
rel = path.relative_to(workspace)
|
||||
except ValueError:
|
||||
continue
|
||||
if any(part.startswith(".") for part in rel.parts):
|
||||
continue
|
||||
try:
|
||||
key = str(rel).replace("\\", "/")
|
||||
result[key] = path.read_text(encoding="utf-8")
|
||||
except (UnicodeDecodeError, OSError):
|
||||
continue
|
||||
return result
|
||||
Reference in New Issue
Block a user