test and fix

Author: 0xRobin

dune_client/api/usage.py

@@ -17,30 +17,31 @@ class UsageAPI(BaseRouter):
 
     def get_usage(
         self,
-        start_date: str,
-        end_date: str,
+        start_date: str | None = None,
+        end_date: str | None = None,
     ) -> UsageResponse:
         """
-        Get credits usage data, overage, number of private queries run, storage, etc.
-        over a specific timeframe.
+        Get credits usage data, billing periods, storage, etc.
         https://docs.dune.com/api-reference/usage/endpoint/get-usage
 
         Args:
-            start_date: Start date for the usage period (ISO format: YYYY-MM-DD)
-            end_date: End date for the usage period (ISO format: YYYY-MM-DD)
+            start_date: Optional start date for the usage period (format: YYYY-MM-DD)
+            end_date: Optional end date for the usage period (format: YYYY-MM-DD)
 
         Returns:
-            UsageResponse containing usage statistics
+            UsageResponse containing usage statistics and billing periods
 
         Requires Plus subscription!
         """
-        params = {
-            "start_date": start_date,
-            "end_date": end_date,
-        }
-        response_json = self._get(
+        payload = {}
+        if start_date:
+            payload["start_date"] = start_date
+        if end_date:
+            payload["end_date"] = end_date
+
+        response_json = self._post(
             route="/usage",
-            params=params,
+            params=payload,
         )
         try:
             return UsageResponse.from_dict(response_json)

dune_client/models.py

@@ -399,48 +399,84 @@ class ClearTableResult(DataClassJsonMixin):
     message: str
 
 
+@dataclass
+class BillingPeriod:
+    """Billing period information with credits and dates"""
+
+    credits_included: float
+    credits_used: float
+    start_date: str
+    end_date: str
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> BillingPeriod:
+        """Constructor from dictionary."""
+        return cls(
+            credits_included=float(data["credits_included"]),
+            credits_used=float(data["credits_used"]),
+            start_date=data["start_date"],
+            end_date=data["end_date"],
+        )
+
+
 @dataclass
 class UsageResponse:
     """
-    Representation of Response from Dune's [GET] Usage endpoint
+    Representation of Response from Dune's [POST] Usage endpoint
     https://docs.dune.com/api-reference/usage/endpoint/get-usage
     """
 
-    credits_used: int
-    overage_credits: int
-    private_query_executions: int
-    storage_bytes: int
+    billing_periods: list[BillingPeriod]
+    bytes_allowed: int
+    bytes_used: int
+    private_dashboards: int
+    private_queries: int
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> UsageResponse:
         """Constructor from dictionary."""
+        billing_periods_data = data.get("billing_periods", [])
         return cls(
-            credits_used=int(data.get("credits_used", 0)),
-            overage_credits=int(data.get("overage_credits", 0)),
-            private_query_executions=int(data.get("private_query_executions", 0)),
-            storage_bytes=int(data.get("storage_bytes", 0)),
+            billing_periods=[BillingPeriod.from_dict(bp) for bp in billing_periods_data],
+            bytes_allowed=int(data.get("bytes_allowed", 0)),
+            bytes_used=int(data.get("bytes_used", 0)),
+            private_dashboards=int(data.get("private_dashboards", 0)),
+            private_queries=int(data.get("private_queries", 0)),
         )
 
 
 @dataclass
 class TableInfo:
     """Information about a single table"""
 
-    namespace: str
-    table_name: str
     full_name: str
-    created_at: str
     is_private: bool
+    created_at: str
+    # Optional fields that may be present
+    table_size_bytes: str | None = None
+    updated_at: str | None = None
+
+    @property
+    def namespace(self) -> str:
+        """Extract namespace from full_name (e.g., 'dune.namespace.table' -> 'namespace')"""
+        parts = self.full_name.split(".")
+        return parts[1] if len(parts) >= 3 else ""
+
+    @property
+    def table_name(self) -> str:
+        """Extract table name from full_name (e.g., 'dune.namespace.table' -> 'table')"""
+        parts = self.full_name.split(".")
+        return parts[2] if len(parts) >= 3 else ""
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> TableInfo:
         """Constructor from dictionary."""
         return cls(
-            namespace=data["namespace"],
-            table_name=data["table_name"],
             full_name=data["full_name"],
-            created_at=data["created_at"],
             is_private=data["is_private"],
+            created_at=data["created_at"],
+            table_size_bytes=data.get("table_size_bytes"),
+            updated_at=data.get("updated_at"),
         )
 
 

tests/e2e/test_client.py

@@ -398,17 +398,26 @@ def test_run_sql(self):
     def test_get_usage(self):
         """Test the get_usage endpoint"""
         dune = DuneClient()
-        usage = dune.get_usage("2024-01-01", "2024-01-31")
+        usage = dune.get_usage("2025-01-01", "2025-02-01")
         # Verify response structure
-        assert hasattr(usage, "credits_used")
-        assert hasattr(usage, "overage_credits")
-        assert hasattr(usage, "private_query_executions")
-        assert hasattr(usage, "storage_bytes")
-        # All should be integers
-        assert isinstance(usage.credits_used, int)
-        assert isinstance(usage.overage_credits, int)
-        assert isinstance(usage.private_query_executions, int)
-        assert isinstance(usage.storage_bytes, int)
+        assert hasattr(usage, "billing_periods")
+        assert hasattr(usage, "bytes_allowed")
+        assert hasattr(usage, "bytes_used")
+        assert hasattr(usage, "private_dashboards")
+        assert hasattr(usage, "private_queries")
+        # Check types
+        assert isinstance(usage.billing_periods, list)
+        assert isinstance(usage.bytes_allowed, int)
+        assert isinstance(usage.bytes_used, int)
+        assert isinstance(usage.private_dashboards, int)
+        assert isinstance(usage.private_queries, int)
+        # If there are billing periods, verify their structure
+        if usage.billing_periods:
+            bp = usage.billing_periods[0]
+            assert hasattr(bp, "credits_included")
+            assert hasattr(bp, "credits_used")
+            assert hasattr(bp, "start_date")
+            assert hasattr(bp, "end_date")
 
     @unittest.skip("Requires Plus subscription and uploaded tables")
     def test_list_tables(self):

tests/unit/test_models.py

@@ -285,56 +285,74 @@ def test_create_table_result_missing_already_existed_field(self):
     def test_usage_response_parsing(self):
         """Test UsageResponse parsing from API response"""
         response_data = {
-            "credits_used": 1000,
-            "overage_credits": 50,
-            "private_query_executions": 25,
-            "storage_bytes": 1024000,
+            "billing_periods": [
+                {
+                    "credits_included": 100.0,
+                    "credits_used": 50.5,
+                    "start_date": "2025-01-01",
+                    "end_date": "2025-02-01",
+                },
+                {
+                    "credits_included": 100.0,
+                    "credits_used": 75.25,
+                    "start_date": "2025-02-01",
+                    "end_date": "2025-03-01",
+                },
+            ],
+            "bytes_allowed": 1024000,
+            "bytes_used": 512000,
+            "private_dashboards": 5,
+            "private_queries": 10,
         }
         result = UsageResponse.from_dict(response_data)
-        assert result.credits_used == 1000
-        assert result.overage_credits == 50
-        assert result.private_query_executions == 25
-        assert result.storage_bytes == 1024000
+        assert len(result.billing_periods) == 2
+        assert result.billing_periods[0].credits_included == 100.0
+        assert result.billing_periods[0].credits_used == 50.5
+        assert result.billing_periods[0].start_date == "2025-01-01"
+        assert result.billing_periods[1].credits_used == 75.25
+        assert result.bytes_allowed == 1024000
+        assert result.bytes_used == 512000
+        assert result.private_dashboards == 5
+        assert result.private_queries == 10
 
     def test_usage_response_parsing_with_missing_fields(self):
         """Test UsageResponse parsing with missing optional fields defaults to 0"""
         response_data = {}
         result = UsageResponse.from_dict(response_data)
-        assert result.credits_used == 0
-        assert result.overage_credits == 0
-        assert result.private_query_executions == 0
-        assert result.storage_bytes == 0
+        assert len(result.billing_periods) == 0
+        assert result.bytes_allowed == 0
+        assert result.bytes_used == 0
+        assert result.private_dashboards == 0
+        assert result.private_queries == 0
 
     def test_table_info_parsing(self):
         """Test TableInfo parsing from API response"""
         response_data = {
-            "namespace": "my_namespace",
-            "table_name": "my_table",
             "full_name": "dune.my_namespace.my_table",
             "created_at": "2024-01-15T10:30:00Z",
             "is_private": True,
+            "table_size_bytes": "1024",
+            "updated_at": "2024-01-16T10:30:00Z",
         }
         result = TableInfo.from_dict(response_data)
+        assert result.full_name == "dune.my_namespace.my_table"
         assert result.namespace == "my_namespace"
         assert result.table_name == "my_table"
-        assert result.full_name == "dune.my_namespace.my_table"
         assert result.created_at == "2024-01-15T10:30:00Z"
         assert result.is_private is True
+        assert result.table_size_bytes == "1024"
+        assert result.updated_at == "2024-01-16T10:30:00Z"
 
     def test_list_tables_response_parsing(self):
         """Test ListTablesResponse parsing from API response"""
         response_data = {
             "tables": [
                 {
-                    "namespace": "namespace1",
-                    "table_name": "table1",
                     "full_name": "dune.namespace1.table1",
                     "created_at": "2024-01-15T10:30:00Z",
                     "is_private": False,
                 },
                 {
-                    "namespace": "namespace2",
-                    "table_name": "table2",
                     "full_name": "dune.namespace2.table2",
                     "created_at": "2024-01-16T11:30:00Z",
                     "is_private": True,
@@ -344,7 +362,11 @@ def test_list_tables_response_parsing(self):
         }
         result = ListTablesResponse.from_dict(response_data)
         assert len(result.tables) == 2
+        assert result.tables[0].full_name == "dune.namespace1.table1"
+        assert result.tables[0].namespace == "namespace1"
         assert result.tables[0].table_name == "table1"
+        assert result.tables[1].full_name == "dune.namespace2.table2"
+        assert result.tables[1].namespace == "namespace2"
         assert result.tables[1].table_name == "table2"
         assert result.next_offset == 100