diff --git a/api/asset.py b/api/asset.py index d7d2b62d..f3261f39 100644 --- a/api/asset.py +++ b/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, diff --git a/tests/test_asset.py b/tests/test_asset.py index 6266e7c7..1b54a4e4 100644 --- a/tests/test_asset.py +++ b/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