feat(api/asset): add new list unassociated assets endpoint

api/asset.py

@@ -468,6 +468,33 @@ def transformer(records: list[Asset]):
     return paginate(query=sql, conn=session, transformer=transformer)
 
 
+@router.get("/unassociated")
+async def list_unassociated_assets(
+    user: viewer_dependency,
+    session: session_dependency,
+) -> CustomPage[AssetResponse]:
+    """
+    List assets that are not associated with any Thing.
+    """
+    sql = (
+        select(Asset)
+        .outerjoin(AssetThingAssociation)
+        .where(AssetThingAssociation.asset_id.is_(None))
+        .order_by(Asset.id)
+    )
+
+    # Signed URLs are generated for thumbnail display on the frontend.
+    # The frontend paginates this endpoint and requests only 10 assets at a time,
+    # which limits GCP IAM calls and keeps signed URL generation manageable.
+    def transformer(records: list[Asset]):
+        from services.gcs_helper import add_signed_url
+
+        bucket = get_storage_bucket()
+        return [add_signed_url(asset, bucket) for asset in records]
+
+    return paginate(query=sql, conn=session, transformer=transformer)
+
+
 @router.get("/{asset_id}")
 async def get_asset(
     user: viewer_dependency,

tests/test_asset.py

@@ -306,6 +306,21 @@ def test_get_assets_thing_id(asset_with_associated_thing, water_well_thing):
         )
 
 
+def test_get_unassociated_assets(asset, asset_with_associated_thing):
+    with patch(
+        "api.asset.get_storage_bucket",
+        return_value=MockStorageBucket(),
+    ):
+        response = client.get("/asset/unassociated")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["total"] == 1
+        assert data["items"][0]["id"] == asset.id
+        expected_signed_url = MockBlob().generate_signed_url()
+        assert data["items"][0]["signed_url"] == expected_signed_url
+        assert data["items"][0]["id"] != asset_with_associated_thing.id
+
+
 def test_get_asset_by_id(asset):
     response = client.get(f"/asset/{asset.id}")
     assert response.status_code == 200