feat: add DataExporter utility class

Author: LifeJiggy

src/gradient/_utils/__init__.py

@@ -35,6 +35,7 @@
     ResponseCache as ResponseCache,
     RateLimiter as RateLimiter,
     BatchProcessor as BatchProcessor,
+    DataExporter as DataExporter,
 )
 from ._compat import (
     get_args as get_args,

src/gradient/_utils/_utils.py

@@ -612,6 +612,109 @@ def is_empty(self) -> bool:
         return len(self._batch) == 0
 
 
+# Data Export Classes
+class DataExporter:
+    """Utility for exporting API response data to JSON/CSV formats."""
+
+    def __init__(self) -> None:
+        """Initialize data exporter."""
+        pass
+
+    def _flatten_response(self, data: Any, prefix: str = "") -> dict[str, Any]:
+        """Flatten nested response data for CSV export."""
+        flattened = {}
+
+        if isinstance(data, dict):
+            for key, value in data.items():
+                new_key = f"{prefix}.{key}" if prefix else key
+                if isinstance(value, (dict, list)):
+                    flattened.update(self._flatten_response(value, new_key))
+                else:
+                    flattened[new_key] = value
+        elif isinstance(data, list):
+            # For lists, create indexed keys
+            for i, item in enumerate(data):
+                new_key = f"{prefix}[{i}]" if prefix else f"[{i}]"
+                if isinstance(item, (dict, list)):
+                    flattened.update(self._flatten_response(item, new_key))
+                else:
+                    flattened[new_key] = item
+        else:
+            flattened[prefix] = data
+
+        return flattened
+
+    def export_json(self, data: Any, file_path: str | None = None, indent: int = 2) -> str | None:
+        """Export data to JSON format.
+
+        Args:
+            data: Data to export
+            file_path: Optional file path to save to
+            indent: JSON indentation level
+
+        Returns:
+            JSON string if no file_path provided, None otherwise
+        """
+        import json
+
+        json_str = json.dumps(data, indent=indent, default=str)
+
+        if file_path:
+            with open(file_path, 'w', encoding='utf-8') as f:
+                f.write(json_str)
+            return None
+
+        return json_str
+
+    def export_csv(self, data: Any, file_path: str | None = None, headers: list[str] | None = None) -> str | None:
+        """Export data to CSV format.
+
+        Args:
+            data: Data to export (list of dicts or single dict)
+            file_path: Optional file path to save to
+            headers: Optional custom headers
+
+        Returns:
+            CSV string if no file_path provided, None otherwise
+        """
+        import csv
+        import io
+
+        # Ensure data is a list
+        if not isinstance(data, list):
+            data = [data]
+
+        # Flatten each item in the data
+        flattened_data = [self._flatten_response(item) for item in data]
+
+        # Determine headers
+        if not headers:
+            all_keys = set()
+            for item in flattened_data:
+                all_keys.update(item.keys())
+            headers = sorted(all_keys)
+
+        # Create CSV content
+        output = io.StringIO() if not file_path else open(file_path, 'w', newline='', encoding='utf-8')
+
+        try:
+            writer = csv.DictWriter(output, fieldnames=headers)
+            writer.writeheader()
+
+            for item in flattened_data:
+                # Fill missing keys with empty strings
+                row = {header: item.get(header, '') for header in headers}
+                writer.writerow(row)
+
+            if not file_path:
+                return output.getvalue()
+            return None
+
+        finally:
+            if file_path:
+                output.close()
+
+
 # API Key Validation Functions
 def validate_api_key(api_key: str | None) -> bool:
     """Validate an API key format.

tests/test_data_exporter.py

@@ -0,0 +1,104 @@
+"""Tests for data export functionality."""
+
+import tempfile
+import os
+import pytest
+from gradient._utils import DataExporter
+
+
+class TestDataExporter:
+    """Test data export functionality."""
+
+    def test_export_json_string(self):
+        """Test JSON export to string."""
+        exporter = DataExporter()
+        data = {"name": "test", "value": 123}
+
+        result = exporter.export_json(data)
+        assert result is not None
+        assert '"name": "test"' in result
+        assert '"value": 123' in result
+
+    def test_export_json_file(self):
+        """Test JSON export to file."""
+        exporter = DataExporter()
+        data = {"items": [1, 2, 3]}
+
+        with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.json') as f:
+            temp_path = f.name
+
+        try:
+            result = exporter.export_json(data, temp_path)
+            assert result is None  # Should return None when saving to file
+
+            # Verify file contents
+            with open(temp_path, 'r') as f:
+                content = f.read()
+                assert '"items": [1, 2, 3]' in content
+
+        finally:
+            os.unlink(temp_path)
+
+    def test_export_csv_string(self):
+        """Test CSV export to string."""
+        exporter = DataExporter()
+        data = [
+            {"name": "Alice", "age": 30},
+            {"name": "Bob", "age": 25}
+        ]
+
+        result = exporter.export_csv(data)
+        assert result is not None
+        assert "name,age" in result
+        assert "Alice,30" in result
+        assert "Bob,25" in result
+
+    def test_export_csv_file(self):
+        """Test CSV export to file."""
+        exporter = DataExporter()
+        data = [{"id": 1, "status": "active"}]
+
+        with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.csv') as f:
+            temp_path = f.name
+
+        try:
+            result = exporter.export_csv(data, temp_path)
+            assert result is None
+
+            # Verify file contents
+            with open(temp_path, 'r') as f:
+                content = f.read()
+                assert "id,status" in content
+                assert "1,active" in content
+
+        finally:
+            os.unlink(temp_path)
+
+    def test_flatten_response(self):
+        """Test response flattening for nested data."""
+        exporter = DataExporter()
+
+        nested_data = {
+            "user": {
+                "name": "John",
+                "profile": {
+                    "age": 30,
+                    "hobbies": ["reading", "coding"]
+                }
+            },
+            "active": True
+        }
+
+        flattened = exporter._flatten_response(nested_data)
+
+        # Check flattened keys exist
+        assert "user.name" in flattened
+        assert "user.profile.age" in flattened
+        assert "user.profile.hobbies[0]" in flattened
+        assert "user.profile.hobbies[1]" in flattened
+        assert "active" in flattened
+
+        # Check values
+        assert flattened["user.name"] == "John"
+        assert flattened["user.profile.age"] == 30
+        assert flattened["active"] is True