docs: properly document the new aggregation pipeline endpoint

Author: dbirman

README.md

@@ -45,9 +45,11 @@ result = response.json()
 ### Query endpoints
 
 - `GET /upgrade-query` — Build a query using the LLM query builder.
-- `POST /retrieve-records` — Run a query against the metadata store.
+- `POST /retrieve-records` — Run a filter query or aggregation pipeline against the metadata store.
 
-Optional query parameters for `/retrieve-records`:
+To run a **filter query**, send a JSON object as the body:
+
+Optional query parameters:
 - `names_only=true` — return only asset names
 - `limit=<int>` — limit number of results (default 0 = no limit)
 - `projection=<json>` — JSON object specifying which fields to include/exclude
@@ -61,6 +63,19 @@ response = requests.post(
 print(response.json())
 ```
 
+To run an **aggregation pipeline**, send a JSON array of pipeline stage dicts as the body. The pipeline is always executed against DocumentDB.
+
+```python
+response = requests.post(
+    "https://metadata-portal.allenneuraldynamics-test.org/retrieve-records",
+    json=[
+        {"$match": {"data_description.project_name": "MyProject"}},
+        {"$group": {"_id": "$subject.subject_id", "count": {"$sum": 1}}},
+    ],
+)
+print(response.json())
+```
+
 ### Chat endpoint
 
 `POST /chat` — Ask a natural-language question about the metadata store. The agent can query records, look up schema info, and summarize results.
@@ -250,9 +265,11 @@ Fields that rename across schema versions (e.g. `session` → `acquisition`, `ri
 ### Query endpoints
 
 - `GET /upgrade-query` — Build a query using the LLM query builder. Pass query string parameters as needed.
-- `POST /retrieve-records` — Run a query. Accepts a JSON object as the request body.
+- `POST /retrieve-records` — Run a filter query or aggregation pipeline against the metadata store.
 
-Optional query parameters for `/retrieve-records`:
+To run a **filter query**, send a JSON object as the body:
+
+Optional query parameters:
 - `names_only=true` — return only asset names
 - `limit=<int>` — limit number of results (default 0 = no limit)
 - `projection=<json>` — JSON object specifying which fields to include/exclude (e.g. `{"subject.subject_id": 1}`)
@@ -266,6 +283,19 @@ response = requests.post(
 print(response.json())
 ```
 
+To run an **aggregation pipeline**, send a JSON array of pipeline stage dicts as the body. The pipeline is always executed against DocumentDB.
+
+```python
+response = requests.post(
+    "https://metadata-portal.allenneuraldynamics-test.org/retrieve-records",
+    json=[
+        {"$match": {"data_description.project_name": "MyProject"}},
+        {"$group": {"_id": "$subject.subject_id", "count": {"$sum": 1}}},
+    ],
+)
+print(response.json())
+```
+
 ### Chat endpoint
 
 `POST /chat` — Ask a natural-language question about the metadata store. The agent can query records, look up schema info, and summarize results.

tests/test_validation.py

@@ -369,5 +369,71 @@ def test_upgrade_by_asset_name_fetch_error(self, mock_retrieve):
         self.assertIn("Failed to fetch record", response.json()["error"])
 
 
+class TestRetrieveRecordsEndpoint(unittest.TestCase):
+
+    @patch("aind_metadata_viz.endpoints.retrieve_records")
+    def test_filter_query(self, mock_retrieve):
+        mock_result = Mock()
+        mock_result.backend = "cache"
+        mock_result.elapsed_seconds = 0.1
+        mock_result.asset_names = ["asset-1"]
+        mock_result.records = [{"name": "asset-1"}]
+        mock_retrieve.return_value = mock_result
+
+        response = client.post("/retrieve-records", json={"subject.subject_id": "123456"})
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertEqual(body["asset_names"], ["asset-1"])
+        self.assertEqual(body["records"], [{"name": "asset-1"}])
+        mock_retrieve.assert_called_once()
+
+    @patch("aind_metadata_viz.endpoints.retrieve_aggregation")
+    def test_aggregation_pipeline(self, mock_aggregate):
+        mock_result = Mock()
+        mock_result.backend = "docdb"
+        mock_result.elapsed_seconds = 0.2
+        mock_result.asset_names = ["asset-2"]
+        mock_result.records = [{"name": "asset-2", "count": 5}]
+        mock_aggregate.return_value = mock_result
+
+        pipeline = [{"$match": {"subject.subject_id": "123456"}}, {"$limit": 5}]
+        response = client.post("/retrieve-records", json=pipeline)
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertEqual(body["backend"], "docdb")
+        self.assertEqual(body["asset_names"], ["asset-2"])
+        self.assertEqual(body["records"], [{"name": "asset-2", "count": 5}])
+        mock_aggregate.assert_called_once_with(pipeline)
+
+    @patch("aind_metadata_viz.endpoints.retrieve_aggregation")
+    def test_aggregation_pipeline_error(self, mock_aggregate):
+        mock_aggregate.side_effect = Exception("pipeline error")
+
+        response = client.post("/retrieve-records", json=[{"$match": {}}])
+        self.assertEqual(response.status_code, 500)
+        self.assertIn("Aggregation execution failed", response.json()["error"])
+
+    @patch("aind_metadata_viz.endpoints.retrieve_records")
+    def test_filter_query_error(self, mock_retrieve):
+        mock_retrieve.side_effect = Exception("connection error")
+
+        response = client.post("/retrieve-records", json={"name": "test"})
+        self.assertEqual(response.status_code, 500)
+        self.assertIn("Query execution failed", response.json()["error"])
+
+    def test_invalid_json(self):
+        response = client.post(
+            "/retrieve-records",
+            content=b"not json",
+            headers={"Content-Type": "application/json"},
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertIn("Invalid JSON format", response.json()["error"])
+
+    def test_invalid_body_type(self):
+        response = client.post("/retrieve-records", json="a string")
+        self.assertEqual(response.status_code, 400)
+
+
 if __name__ == "__main__":
     unittest.main()