fix it

Author: shntu

roboflow/core/project.py

@@ -869,3 +869,68 @@ def create_annotation_job(
                 raise RuntimeError(f"Failed to create annotation job: {response.text}")
 
         return response.json()
+
+    def get_batches(self) -> Dict:
+        """
+        Get a list of all batches in the project.
+
+        Returns:
+            Dict: A dictionary containing the list of batches
+
+        Example:
+            >>> import roboflow
+
+            >>> rf = roboflow.Roboflow(api_key="YOUR_API_KEY")
+
+            >>> project = rf.workspace().project("PROJECT_ID")
+
+            >>> batches = project.get_batches()
+        """
+        url = f"{API_URL}/{self.__workspace}/{self.__project_name}/batches?api_key={self.__api_key}"
+
+        response = requests.get(url)
+
+        if response.status_code != 200:
+            try:
+                error_data = response.json()
+                if "error" in error_data:
+                    raise RuntimeError(error_data["error"])
+                raise RuntimeError(response.text)
+            except ValueError:
+                raise RuntimeError(f"Failed to get batches: {response.text}")
+
+        return response.json()
+
+    def get_batch(self, batch_id: str) -> Dict:
+        """
+        Get information for a specific batch in the project.
+
+        Args:
+            batch_id (str): The ID of the batch to retrieve
+
+        Returns:
+            Dict: A dictionary containing the batch details
+
+        Example:
+            >>> import roboflow
+
+            >>> rf = roboflow.Roboflow(api_key="YOUR_API_KEY")
+
+            >>> project = rf.workspace().project("PROJECT_ID")
+
+            >>> batch = project.get_batch("batch123")
+        """
+        url = f"{API_URL}/{self.__workspace}/{self.__project_name}/batches/{batch_id}?api_key={self.__api_key}"
+
+        response = requests.get(url)
+
+        if response.status_code != 200:
+            try:
+                error_data = response.json()
+                if "error" in error_data:
+                    raise RuntimeError(error_data["error"])
+                raise RuntimeError(response.text)
+            except ValueError:
+                raise RuntimeError(f"Failed to get batch {batch_id}: {response.text}")
+
+        return response.json()

tests/test_project.py

@@ -454,3 +454,82 @@ def test_project_upload_dataset(self):
             finally:
                 for mock in mocks.values():
                     mock.stop()
+        
+    def test_get_batches_success(self):
+        expected_url = f"{API_URL}/{WORKSPACE_NAME}/{PROJECT_NAME}/batches?api_key={ROBOFLOW_API_KEY}"
+        mock_response = {
+            "batches": [
+                {
+                    "id": "batch-1",
+                    "name": "Batch 1",
+                    "created": 1616161616,
+                    "images": 10,
+                },
+                {
+                    "id": "batch-2",
+                    "name": "Batch 2",
+                    "created": 1616161617,
+                    "images": 5,
+                }
+            ]
+        }
+
+        responses.add(responses.GET, expected_url, json=mock_response, status=200)
+
+        batches = self.project.get_batches()
+
+        self.assertIsInstance(batches, dict)
+        self.assertIn("batches", batches)
+        self.assertEqual(len(batches["batches"]), 2)
+        self.assertEqual(batches["batches"][0]["id"], "batch-1")
+        self.assertEqual(batches["batches"][1]["id"], "batch-2")
+
+    def test_get_batches_error(self):
+        expected_url = f"{API_URL}/{WORKSPACE_NAME}/{PROJECT_NAME}/batches?api_key={ROBOFLOW_API_KEY}"
+        error_response = {"error": "Cannot retrieve batches"}
+
+        responses.add(responses.GET, expected_url, json=error_response, status=404)
+
+        with self.assertRaises(RuntimeError) as context:
+            self.project.get_batches()
+
+        self.assertEqual(str(context.exception), "Cannot retrieve batches")
+
+    def test_get_batch_success(self):
+        batch_id = "batch-123"
+        expected_url = f"{API_URL}/{WORKSPACE_NAME}/{PROJECT_NAME}/batches/{batch_id}?api_key={ROBOFLOW_API_KEY}"
+        mock_response = {
+            "batch": {
+                "id": batch_id,
+                "name": "My Test Batch",
+                "created": 1616161616,
+                "images": 25,
+                "metadata": {
+                    "source": "API Upload",
+                    "type": "test"
+                }
+            }
+        }
+
+        responses.add(responses.GET, expected_url, json=mock_response, status=200)
+
+        batch = self.project.get_batch(batch_id)
+
+        self.assertIsInstance(batch, dict)
+        self.assertIn("batch", batch)
+        self.assertEqual(batch["batch"]["id"], batch_id)
+        self.assertEqual(batch["batch"]["name"], "My Test Batch")
+        self.assertEqual(batch["batch"]["images"], 25)
+        self.assertIn("metadata", batch["batch"])
+
+    def test_get_batch_error(self):
+        batch_id = "nonexistent-batch"
+        expected_url = f"{API_URL}/{WORKSPACE_NAME}/{PROJECT_NAME}/batches/{batch_id}?api_key={ROBOFLOW_API_KEY}"
+        error_response = {"error": "Batch not found"}
+
+        responses.add(responses.GET, expected_url, json=error_response, status=404)
+
+        with self.assertRaises(RuntimeError) as context:
+            self.project.get_batch(batch_id)
+
+        self.assertEqual(str(context.exception), "Batch not found")