Add asset file path response fields

Amp-Thread-ID: https://ampcode.com/threads/T-019e4307-dd77-7709-b9f4-46bb79dcf58a
Co-authored-by: Amp <amp@ampcode.com>

app/assets/api/routes.py

@@ -39,6 +39,7 @@ from app.assets.services import (
     update_asset_metadata,
     upload_from_temp_path,
 )
+from app.assets.services.path_utils import compute_paths_for_response
 from app.assets.services.tagging import list_tag_histogram
 
 ROUTES = web.RouteTableDef()
@@ -160,9 +161,16 @@ def _build_asset_response(result: schemas.AssetDetailResult | schemas.UploadResu
             preview_url = None
     else:
         preview_url = _build_preview_url_from_view(result.tags, result.ref.user_metadata)
+    if result.ref.file_path:
+        paths = compute_paths_for_response(result.ref.file_path)
+        file_path, display_name = paths if paths else (None, None)
+    else:
+        file_path, display_name = None, None
     return schemas_out.Asset(
         id=result.ref.id,
         name=result.ref.name,
+        file_path=file_path,
+        display_name=display_name,
         asset_hash=result.asset.hash if result.asset else None,
         size=int(result.asset.size_bytes) if result.asset else None,
         mime_type=result.asset.mime_type if result.asset else None,

app/assets/api/schemas_out.py

@@ -10,6 +10,8 @@ class Asset(BaseModel):
 
     id: str
     name: str
+    file_path: str | None = None
+    display_name: str | None = None
     asset_hash: str | None = None
     size: int | None = None
     mime_type: str | None = None

app/assets/services/path_utils.py

@@ -8,6 +8,8 @@ from app.assets.helpers import normalize_tags
 
 _NON_MODEL_FOLDER_NAMES = frozenset({"custom_nodes"})
 
+RootCategory = Literal["input", "output", "temp", "models"]
+
 
 def get_comfy_models_folders() -> list[tuple[str, list[str]]]:
     """Build list of (folder_name, base_paths[]) for all model locations.
@@ -65,35 +67,109 @@ def validate_path_within_base(candidate: str, base: str) -> None:
         raise ValueError("destination escapes base directory")
 
 
-def compute_relative_filename(file_path: str) -> str | None:
+def compute_paths_for_response(
-    """
+    file_path: str,
-    Return the model's path relative to the last well-known folder (the model category),
+) -> tuple[str, str | None] | None:
-    using forward slashes, eg:
+    """Compute (file_path, display_name) for an Asset response.
-      /.../models/checkpoints/flux/123/flux.safetensors -> "flux/123/flux.safetensors"
-      /.../models/text_encoders/clip_g.safetensors -> "clip_g.safetensors"
 
-    For non-model paths, returns None.
+    `file_path` is a logical locator under the asset namespace: `<root>/<rel>`
+    for input/output/temp assets and `<root>/<bucket>/<rel>` for model assets.
+    `display_name` is the path below that root or model bucket, suitable for UI
+    labels. Returns None when the absolute path is not under a known asset root.
     """
     try:
-        root_category, rel_path = get_asset_category_and_relative_path(file_path)
+        root, bucket, rel = get_asset_root_bucket_and_filepath(file_path)
     except ValueError:
         return None
 
-    p = Path(rel_path)
+    display_name = rel or None
-    parts = [seg for seg in p.parts if seg not in (".", "..", p.anchor)]
+    if bucket is None:
-    if not parts:
+        response_file_path = f"{root}/{rel}" if rel else root
-        return None
+    else:
+        response_file_path = f"{root}/{bucket}/{rel}" if rel else f"{root}/{bucket}"
+    return response_file_path, display_name
 
-    if root_category == "models":
+
-        # parts[0] is the category ("checkpoints", "vae", etc) – drop it
+def compute_display_name(file_path: str) -> str | None:
-        inside = parts[1:] if len(parts) > 1 else [parts[0]]
+    """Return the asset's `display_name`, or None for unknown paths."""
-        return "/".join(inside)
+    result = compute_paths_for_response(file_path)
-    return "/".join(parts)  # input/output: keep all parts
+    return result[1] if result else None
+
+
+def compute_file_path(file_path: str) -> str | None:
+    """Return the asset's logical `file_path`, or None for unknown paths."""
+    result = compute_paths_for_response(file_path)
+    return result[0] if result else None
+
+
+def compute_relative_filename(file_path: str) -> str | None:
+    """
+    Return the path relative to the asset root or model category, using forward slashes, eg:
+      /.../models/checkpoints/flux/123/flux.safetensors -> "flux/123/flux.safetensors"
+      /.../models/text_encoders/clip_g.safetensors -> "clip_g.safetensors"
+      /.../input/sub/image.png -> "sub/image.png"
+
+    For unknown paths, returns None.
+    """
+    return compute_display_name(file_path)
+
+
+def get_asset_root_bucket_and_filepath(
+    file_path: str,
+) -> tuple[RootCategory, str | None, str]:
+    """Decompose an absolute path into (root, bucket, path-under-bucket).
+
+    `bucket` is only set for model assets. The returned relative path always
+    uses `/` separators and is empty when the path is exactly the matched root.
+
+    Raises:
+        ValueError: path does not belong to any known root.
+    """
+    fp_abs = os.path.abspath(file_path)
+
+    def _check_is_within(child: str, parent: str) -> bool:
+        return Path(child).is_relative_to(parent)
+
+    def _compute_relative(child: str, parent: str) -> str:
+        # Normalize relative path, stripping any leading ".." components
+        # by anchoring to root (os.sep) then computing relpath back from it.
+        rel = os.path.relpath(
+            os.path.join(os.sep, os.path.relpath(child, parent)), os.sep
+        )
+        return "" if rel == "." else rel.replace(os.sep, "/")
+
+    for root_tag, getter in (
+        ("input", folder_paths.get_input_directory),
+        ("output", folder_paths.get_output_directory),
+        ("temp", folder_paths.get_temp_directory),
+    ):
+        base = os.path.abspath(getter())
+        if _check_is_within(fp_abs, base):
+            return root_tag, None, _compute_relative(fp_abs, base)
+
+    # models: check deepest matching base to avoid ambiguity.
+    best: tuple[int, str, str] | None = None
+    for bucket, bases in get_comfy_models_folders():
+        for b in bases:
+            base_abs = os.path.abspath(b)
+            if not _check_is_within(fp_abs, base_abs):
+                continue
+            cand = (len(base_abs), bucket, _compute_relative(fp_abs, base_abs))
+            if best is None or cand[0] > best[0]:
+                best = cand
+
+    if best is not None:
+        _, bucket, rel_inside = best
+        return "models", bucket, rel_inside
+
+    raise ValueError(
+        f"Path is not within input, output, temp, or configured model bases: {file_path}"
+    )
 
 
 def get_asset_category_and_relative_path(
     file_path: str,
-) -> tuple[Literal["input", "output", "temp", "models"], str]:
+) -> tuple[RootCategory, str]:
     """Determine which root category a file path belongs to.
 
     Categories:

openapi.yaml

@@ -6329,6 +6329,16 @@ components:
         name:
           type: string
           description: Name of the asset file
+        file_path:
+          type: string
+          nullable: true
+          x-runtime: [cloud, local]
+          description: "Logical asset locator under the namespace root. Not a unique reference key; use `id` for identity."
+        display_name:
+          type: string
+          nullable: true
+          x-runtime: [cloud, local]
+          description: "Path under the asset's category or root, for human-facing labels."
         hash:
           type: string
           nullable: true
@@ -8109,4 +8119,4 @@ components:
           items:
             $ref: "#/components/schemas/TaskEntry"
         pagination:
-          $ref: "#/components/schemas/PaginationInfo"
+          $ref: "#/components/schemas/PaginationInfo"

tests-unit/assets_test/services/test_path_utils.py

@@ -6,7 +6,11 @@ from unittest.mock import patch
 
 import pytest
 
-from app.assets.services.path_utils import get_asset_category_and_relative_path
+from app.assets.services.path_utils import (
+    compute_display_name,
+    compute_file_path,
+    get_asset_category_and_relative_path,
+)
 
 
 @pytest.fixture
@@ -79,3 +83,27 @@ class TestGetAssetCategoryAndRelativePath:
     def test_unknown_path_raises(self, fake_dirs):
         with pytest.raises(ValueError, match="not within"):
             get_asset_category_and_relative_path("/some/random/path.png")
+
+
+class TestResponsePaths:
+    def test_input_file_path_and_display_name_include_subfolder(self, fake_dirs):
+        sub = fake_dirs["input"] / "some" / "folder"
+        sub.mkdir(parents=True)
+        f = sub / "image.png"
+        f.touch()
+
+        assert compute_file_path(str(f)) == "input/some/folder/image.png"
+        assert compute_display_name(str(f)) == "some/folder/image.png"
+
+    def test_model_file_path_includes_bucket_display_name_drops_it(self, fake_dirs):
+        sub = fake_dirs["models"] / "flux"
+        sub.mkdir()
+        f = sub / "model.safetensors"
+        f.touch()
+
+        assert compute_file_path(str(f)) == "models/checkpoints/flux/model.safetensors"
+        assert compute_display_name(str(f)) == "flux/model.safetensors"
+
+    def test_unknown_path_returns_none(self, fake_dirs):
+        assert compute_file_path("/some/random/path.png") is None
+        assert compute_display_name("/some/random/path.png") is None

tests-unit/assets_test/test_uploads.py

@@ -5,6 +5,8 @@ from concurrent.futures import ThreadPoolExecutor
 import requests
 import pytest
 
+from helpers import get_asset_filename
+
 
 def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, make_asset_bytes):
     name = "dup_a.safetensors"
@@ -63,6 +65,14 @@ def test_upload_fastpath_from_existing_hash_no_file(http: requests.Session, api_
     assert r2.status_code == 200, b2  # fast path returns 200 with created_new == False
     assert b2["created_new"] is False
     assert b2["asset_hash"] == h
+    assert b2.get("file_path") is None
+    assert b2.get("display_name") is None
+
+    rg = http.get(f"{api_base}/api/assets/{b2['id']}", timeout=120)
+    detail = rg.json()
+    assert rg.status_code == 200, detail
+    assert detail.get("file_path") is None
+    assert detail.get("display_name") is None
 
 
 def test_upload_fastpath_with_known_hash_and_file(
@@ -107,6 +117,54 @@ def test_upload_multiple_tags_fields_are_merged(http: requests.Session, api_base
     assert {"models", "checkpoints", "unit-tests", "alpha"}.issubset(tags)
 
 
+@pytest.mark.parametrize(
+    ("tags", "extension", "expected_prefix", "expected_display_prefix"),
+    [
+        (["input", "unit-tests"], ".png", "input", ""),
+        (["models", "checkpoints", "unit-tests"], ".safetensors", "models/checkpoints", ""),
+    ],
+)
+def test_upload_response_includes_file_path_and_display_name(
+    tags: list[str],
+    extension: str,
+    expected_prefix: str,
+    expected_display_prefix: str,
+    http: requests.Session,
+    api_base: str,
+    asset_factory,
+    make_asset_bytes,
+):
+    scope = f"response-paths-{uuid.uuid4().hex[:6]}"
+    scoped_tags = [*tags, scope]
+    name = f"asset_response_path{extension}"
+
+    created = asset_factory(name, scoped_tags, {}, make_asset_bytes(name, 1024))
+    stored_filename = get_asset_filename(created["asset_hash"], extension)
+    expected_suffix = f"unit-tests/{scope}/{stored_filename}"
+    expected_file_path = f"{expected_prefix}/{expected_suffix}"
+    expected_display_name = f"{expected_display_prefix}{expected_suffix}"
+
+    assert created["file_path"] == expected_file_path
+    assert created["display_name"] == expected_display_name
+
+    detail_r = http.get(f"{api_base}/api/assets/{created['id']}", timeout=120)
+    detail = detail_r.json()
+    assert detail_r.status_code == 200, detail
+    assert detail["file_path"] == expected_file_path
+    assert detail["display_name"] == expected_display_name
+
+    list_r = http.get(
+        api_base + "/api/assets",
+        params={"include_tags": f"unit-tests,{scope}", "limit": "50"},
+        timeout=120,
+    )
+    listed = list_r.json()
+    assert list_r.status_code == 200, listed
+    match = next(a for a in listed["assets"] if a["id"] == created["id"])
+    assert match["file_path"] == expected_file_path
+    assert match["display_name"] == expected_display_name
+
+
 @pytest.mark.parametrize("root", ["input", "output"])
 def test_concurrent_upload_identical_bytes_different_names(
     root: str,