Add rename functionality for files and directories

tests/web/test_main.py

@@ -138,6 +138,81 @@ def test_update_file(project_tmp_path: Path) -> None:
     assert (project_tmp_path / "foo.txt").read_text() == "baz"
 
 
+def test_rename_file(project_tmp_path: Path) -> None:
+    txt_file = project_tmp_path / "foo.txt"
+    txt_file.write_text("bar")
+
+    response = client.post("/api/files/foo.txt", json={"new_path": "baz.txt"})
+    assert response.status_code == 200
+    assert response.json() == {
+        "name": "baz.txt",
+        "path": "baz.txt",
+        "extension": ".txt",
+        "is_supported": False,
+        "content": "bar",
+    }
+    assert not txt_file.exists()
+    assert (project_tmp_path / "baz.txt").read_text() == "bar"
+
+
+def test_rename_and_update_file(project_tmp_path: Path) -> None:
+    txt_file = project_tmp_path / "foo.txt"
+    txt_file.write_text("bar")
+
+    response = client.post(
+        "/api/files/foo.txt", json={"content": "hello world", "new_path": "baz.txt"}
+    )
+    assert response.status_code == 200
+    assert response.json() == {
+        "name": "baz.txt",
+        "path": "baz.txt",
+        "extension": ".txt",
+        "is_supported": False,
+        "content": "hello world",
+    }
+    assert not txt_file.exists()
+    assert (project_tmp_path / "baz.txt").read_text() == "hello world"
+
+
+def test_rename_file_not_found(project_tmp_path: Path) -> None:
+    response = client.post("/api/files/foo.txt", json={"new_path": "baz.txt"})
+    assert response.status_code == 404
+
+
+def test_rename_file_already_exists(project_tmp_path: Path) -> None:
+    foo_file = project_tmp_path / "foo.txt"
+    foo_file.write_text("foo")
+    bar_file = project_tmp_path / "bar.txt"
+    bar_file.write_text("bar")
+
+    response = client.post("/api/files/foo.txt", json={"new_path": "bar.txt"})
+    assert response.status_code == 200
+    assert response.json() == {
+        "name": "bar.txt",
+        "path": "bar.txt",
+        "extension": ".txt",
+        "is_supported": False,
+        "content": "foo",
+    }
+    assert not foo_file.exists()
+
+
+def test_rename_file_to_existing_directory(project_tmp_path: Path) -> None:
+    foo_file = project_tmp_path / "foo.txt"
+    foo_file.touch()
+    existing_dir = project_tmp_path / "existing_dir"
+    existing_dir.mkdir()
+
+    response = client.post("/api/files/foo.txt", json={"new_path": "existing_dir"})
+    assert response.status_code == 422
+    assert foo_file.exists()
+
+
+def test_write_file_empty_body(project_tmp_path: Path) -> None:
+    response = client.post("/api/files/foo.txt", json={})
+    assert response.status_code == 404
+
+
 def test_delete_file(project_tmp_path: Path) -> None:
     txt_file = project_tmp_path / "foo.txt"
     txt_file.write_text("bar")
@@ -156,6 +231,7 @@ def test_create_directory(project_tmp_path: Path) -> None:
     response = client.post("/api/directories/new_dir")
     assert response.status_code == 200
     assert (project_tmp_path / "new_dir").exists()
+    assert response.json() == {"directories": [], "files": [], "name": "new_dir", "path": "new_dir"}
 
 
 def test_create_directory_already_exists(project_tmp_path: Path) -> None:
@@ -167,6 +243,64 @@ def test_create_directory_already_exists(project_tmp_path: Path) -> None:
     assert response.json() == {"detail": "Directory already exists"}
 
 
+def test_rename_directory(project_tmp_path: Path) -> None:
+    new_dir = project_tmp_path / "new_dir"
+    new_dir.mkdir()
+
+    response = client.post("/api/directories/new_dir", json={"new_path": "renamed_dir"})
+    assert response.status_code == 200
+    assert not new_dir.exists()
+    assert (project_tmp_path / "renamed_dir").exists()
+    assert response.json() == {
+        "directories": [],
+        "files": [],
+        "name": "renamed_dir",
+        "path": "renamed_dir",
+    }
+
+
+def test_rename_directory_already_exists_empty(project_tmp_path: Path) -> None:
+    new_dir = project_tmp_path / "new_dir"
+    new_dir.mkdir()
+    existing_dir = project_tmp_path / "renamed_dir"
+    existing_dir.mkdir()
+
+    response = client.post("/api/directories/new_dir", json={"new_path": "renamed_dir"})
+    assert response.status_code == 200
+    assert not new_dir.exists()
+    assert (project_tmp_path / "renamed_dir").exists()
+    assert response.json() == {
+        "directories": [],
+        "files": [],
+        "name": "renamed_dir",
+        "path": "renamed_dir",
+    }
+
+
+def test_rename_directory_already_exists_not_empty(project_tmp_path: Path) -> None:
+    new_dir = project_tmp_path / "new_dir"
+    new_dir.mkdir()
+    existing_dir = project_tmp_path / "renamed_dir"
+    existing_dir.mkdir()
+    existing_file = existing_dir / "foo.txt"
+    existing_file.touch()
+
+    response = client.post("/api/directories/new_dir", json={"new_path": "renamed_dir"})
+    assert response.status_code == 422
+    assert new_dir.exists()
+
+
+def test_rename_directory_to_existing_file(project_tmp_path: Path) -> None:
+    new_dir = project_tmp_path / "new_dir"
+    new_dir.mkdir()
+    existing_file = project_tmp_path / "foo.txt"
+    existing_file.touch()
+
+    response = client.post("/api/directories/new_dir", json={"new_path": "foo.txt"})
+    assert response.status_code == 422
+    assert new_dir.exists()
+
+
 def test_delete_directory(project_tmp_path: Path) -> None:
     new_dir = project_tmp_path / "new_dir"
     new_dir.mkdir()

web/server/api/endpoints/directories.py

@@ -1,23 +1,32 @@
 import os
 import shutil
+import typing as t
 
-from fastapi import APIRouter, Depends, HTTPException, Response, status
+from fastapi import APIRouter, Body, Depends, HTTPException, Response, status
 from starlette.status import HTTP_404_NOT_FOUND, HTTP_422_UNPROCESSABLE_ENTITY
 
+from sqlmesh.core.context import Context
 from web.server import models
-from web.server.settings import Settings, get_settings
-from web.server.utils import validate_path
+from web.server.settings import Settings, get_context, get_settings
+from web.server.utils import replace_file, validate_path
 
 router = APIRouter()
 
 
 @router.post("/{path:path}", response_model=models.Directory)
-async def create_directory(
+async def write_directory(
     response: Response,
     path: str = Depends(validate_path),
+    new_path: t.Optional[str] = Body(None, embed=True),
     settings: Settings = Depends(get_settings),
+    context: Context = Depends(get_context),
 ) -> models.Directory:
-    """Create a directory."""
+    """Create or rename a directory."""
+    if new_path:
+        validate_path(new_path, context)
+        replace_file(settings.project_path / path, settings.project_path / new_path)
+        return models.Directory(name=os.path.basename(new_path), path=new_path)
+
     try:
         (settings.project_path / path).mkdir(parents=True)
         return models.Directory(name=os.path.basename(path), path=path)

web/server/api/endpoints/files.py

@@ -10,7 +10,7 @@
 from sqlmesh.core.context import Context
 from web.server import models
 from web.server.settings import Settings, get_context, get_settings
-from web.server.utils import validate_path
+from web.server.utils import replace_file, validate_path
 
 router = APIRouter()
 
@@ -77,14 +77,30 @@ def get_file(
 
 @router.post("/{path:path}", response_model=models.File)
 async def write_file(
-    content: str = Body(embed=True),
+    content: t.Optional[str] = Body(None, embed=True),
+    new_path: t.Optional[str] = Body(None, embed=True),
     path: str = Depends(validate_path),
     settings: Settings = Depends(get_settings),
+    context: Context = Depends(get_context),
 ) -> models.File:
-    """Create or update a file."""
-    with open(settings.project_path / path, "w", encoding="utf-8") as f:
-        f.write(content)
-    return models.File(name=os.path.basename(path), path=path, content=content)
+    """Create, update, or rename a file."""
+    path_or_new_path = path
+    if new_path:
+        path_or_new_path = validate_path(new_path, context)
+        replace_file(settings.project_path / path, settings.project_path / path_or_new_path)
+
+    if content:
+        with open(settings.project_path / path_or_new_path, "w", encoding="utf-8") as f:
+            f.write(content)
+    else:
+        try:
+            with open(settings.project_path / path_or_new_path) as f:
+                content = f.read()
+        except FileNotFoundError:
+            raise HTTPException(status_code=HTTP_404_NOT_FOUND)
+    return models.File(
+        name=os.path.basename(path_or_new_path), path=path_or_new_path, content=content
+    )
 
 
 @router.delete("/{path:path}")

web/server/utils.py

@@ -1,8 +1,10 @@
 import asyncio
+import traceback
 import typing as t
+from pathlib import Path
 
 from fastapi import Depends, HTTPException
-from starlette.status import HTTP_404_NOT_FOUND
+from starlette.status import HTTP_404_NOT_FOUND, HTTP_422_UNPROCESSABLE_ENTITY
 
 from sqlmesh.core.context import Context
 from web.server.settings import get_context
@@ -29,3 +31,16 @@ def validate_path(
         raise HTTPException(status_code=HTTP_404_NOT_FOUND)
 
     return path
+
+
+def replace_file(src: Path, dst: Path) -> None:
+    """Move a file or directory at src to dst."""
+    if src != dst:
+        try:
+            src.replace(dst)
+        except FileNotFoundError:
+            raise HTTPException(status_code=HTTP_404_NOT_FOUND)
+        except OSError:
+            raise HTTPException(
+                status_code=HTTP_422_UNPROCESSABLE_ENTITY, detail=traceback.format_exc()
+            )