feat: [DPS-44447] Add endpoint to expose streams limit per account

Author: odyszews

linode_api4/groups/monitor.py

@@ -22,6 +22,7 @@
     AkamaiObjectStorageLogsDestinationDetails,
     CustomHTTPSLogsDestinationDetails,
     LogsStreamDetails,
+    LogsStreamQuota,
 )
 
 __all__ = [
@@ -625,3 +626,20 @@ def stream_create(
             )
 
         return LogsStream(self.client, result["id"], result)
+
+    def streams_quotas(self, *filters) -> PaginatedList:
+        """
+        Retrieve quota definitions available for the authenticated account's streams.
+
+        This reads from ``/monitor/streams/quotas`` on the API v4 endpoint.
+
+        :param filters: Any number of filters to apply to this query.
+                        See :doc:`Filtering Collections</linode_api4/objects/filtering>`
+                        for more details on filtering.
+
+        :returns: A paginated list of stream quota definitions.
+        :rtype: PaginatedList[LogsStreamQuota]
+        """
+        return self.client._get_and_filter(
+            LogsStreamQuota, *filters, endpoint="/monitor/streams/quotas"
+        )

linode_api4/objects/monitor.py

@@ -40,6 +40,7 @@
     "LogsStreamStatus",
     "LogsStreamDetails",
     "LogsStreamDestination",
+    "LogsStreamQuota",
 ]
 
 
@@ -904,3 +905,16 @@ def history(self):
             "{}/history".format(LogsStream.api_endpoint.format(id=self.id)),
             LogsStreamHistory,
         )
+
+
+@dataclass
+class LogsStreamQuota(JSONObject):
+    """
+    Represents a quota definition for logs streams.
+    """
+
+    quota_id: str = ""
+    quota_name: str = ""
+    quota_limit: int = 0
+    quota_type: str = ""
+    description: str = ""

test/fixtures/monitor_streams_quotas.json

@@ -0,0 +1,21 @@
+{
+  "data": [
+    {
+      "quota_id": "aclp_audit_logs_streams",
+      "quota_name": "Number of Audit Logs Streams",
+      "description": "Current number of audit logs streams per account",
+      "quota_limit": 5,
+      "quota_type": "logs_streams_aclp_audit_logs_streams"
+    },
+    {
+      "quota_id": "aclp_lke_audit_logs_streams",
+      "quota_name": "Number of LKE Audit Logs Streams",
+      "description": "Current number of LKE audit logs streams per account",
+      "quota_limit": 10,
+      "quota_type": "logs_streams_aclp_lke_audit_logs_streams"
+    }
+  ],
+  "results": 2,
+  "page": 1,
+  "pages": 1
+}

test/unit/objects/monitor_test.py

@@ -16,6 +16,7 @@
     DestinationAuthentication,
     LogsDestinationDetailsBase,
     LogsStreamDetails,
+    LogsStreamQuota,
     LogsStreamType,
 )
 
@@ -653,6 +654,41 @@ def test_delete_stream(self):
 
         self.assertEqual(m.call_url, "/monitor/streams/1")
 
+    def test_streams_quotas(self):
+        """
+        Test that streams_quotas returns a paginated list of quota objects.
+        """
+        url = "/monitor/streams/quotas"
+        with self.mock_get(url) as m:
+            result = self.client.monitor.streams_quotas()
+
+        self.assertEqual(m.call_url, url)
+        self.assertEqual(len(result), 2)
+        self.assertIsInstance(result[0], LogsStreamQuota)
+        self.assertEqual(result[0].quota_id, "aclp_audit_logs_streams")
+        self.assertEqual(result[0].quota_name, "Number of Audit Logs Streams")
+        self.assertEqual(
+            result[0].description,
+            "Current number of audit logs streams per account",
+        )
+        self.assertEqual(result[0].quota_limit, 5)
+        self.assertEqual(
+            result[0].quota_type, "logs_streams_aclp_audit_logs_streams"
+        )
+        self.assertIsInstance(result[1], LogsStreamQuota)
+        self.assertEqual(result[1].quota_id, "aclp_lke_audit_logs_streams")
+        self.assertEqual(
+            result[1].quota_name, "Number of LKE Audit Logs Streams"
+        )
+        self.assertEqual(
+            result[1].description,
+            "Current number of LKE audit logs streams per account",
+        )
+        self.assertEqual(result[1].quota_limit, 10)
+        self.assertEqual(
+            result[1].quota_type, "logs_streams_aclp_lke_audit_logs_streams"
+        )
+
 
 class LkeAuditLogsStreamTest(ClientBaseCase):
     """