Add HTTP Method as a Parameter for APIs that support POST (#302)

* Add HTTP Method as a Parameter for APIs that support POST

* Update class docstring

* Add test cases for the method parameter

---------

Co-authored-by: Lam Tran <[email protected]>

Author: s3714110

prometheus_api_client/prometheus_connect.py

@@ -41,6 +41,8 @@ class PrometheusConnect:
         Example: {"http_proxy": "<ip_address/hostname:port>", "https_proxy": "<ip_address/hostname:port>"}
     :param session (Optional) Custom requests.Session to enable complex HTTP configuration
     :param timeout: (Optional) A timeout (in seconds) applied to all requests
+    :param method: (Optional) (str) HTTP Method (GET or POST) to use for Query APIs that allow POST 
+        (/query, /query_range and /labels). Use POST for large and complex queries. Default is GET.
     """
 
     def __init__(
@@ -53,6 +55,7 @@ def __init__(
         proxy: dict = None,
         session: Session = None,
         timeout: int = None,
+        method: str = "GET"
     ):
         """Functions as a Constructor for the class PrometheusConnect."""
         if url is None:
@@ -64,6 +67,15 @@ def __init__(
         self._all_metrics = None
         self._timeout = timeout
 
+        if not isinstance(method, str):
+            raise TypeError("Method must be a string")
+        
+        method = method.upper()
+        if method not in {"GET", "POST"}:
+            raise ValueError("Method can only be GET or POST")
+
+        self._method = method
+
         if retry is None:
             retry = Retry(
                 total=MAX_REQUEST_RETRIES,
@@ -91,8 +103,9 @@ def check_prometheus_connection(self, params: dict = None) -> bool:
             sent along with the API request.
         :returns: (bool) True if the endpoint can be reached, False if cannot be reached.
         """
-        response = self._session.get(
-            "{0}/".format(self.url),
+        response = self._session.request(
+            method="GET",
+            url="{0}/".format(self.url),
             verify=self._session.verify,
             headers=self.headers,
             params=params,
@@ -165,8 +178,9 @@ def get_label_names(self, params: dict = None):
             (PrometheusApiClientException) Raises in case of non 200 response status code
         """
         params = params or {}
-        response = self._session.get(
-            "{0}/api/v1/labels".format(self.url),
+        response = self._session.request(
+            method=self._method,
+            url="{0}/api/v1/labels".format(self.url),
             verify=self._session.verify,
             headers=self.headers,
             params=params,
@@ -196,8 +210,9 @@ def get_label_values(self, label_name: str, params: dict = None):
             (PrometheusApiClientException) Raises in case of non 200 response status code
         """
         params = params or {}
-        response = self._session.get(
-            "{0}/api/v1/label/{1}/values".format(self.url, label_name),
+        response = self._session.request(
+            method="GET",
+            url="{0}/api/v1/label/{1}/values".format(self.url, label_name),
             verify=self._session.verify,
             headers=self.headers,
             params=params,
@@ -248,8 +263,9 @@ def get_current_metric_value(
             query = metric_name
 
         # using the query API to get raw data
-        response = self._session.get(
-            "{0}/api/v1/query".format(self.url),
+        response = self._session.request(
+            method=self._method,
+            url="{0}/api/v1/query".format(self.url),
             params={**{"query": query}, **params},
             verify=self._session.verify,
             headers=self.headers,
@@ -335,8 +351,9 @@ def get_metric_range_data(
                 chunk_seconds = end - start
 
             # using the query API to get raw data
-            response = self._session.get(
-                "{0}/api/v1/query".format(self.url),
+            response = self._session.request(
+                method=self._method,
+                url="{0}/api/v1/query".format(self.url),
                 params={
                     **{
                         "query": query + "[" + str(chunk_seconds) + "s" + "]",
@@ -443,8 +460,9 @@ def custom_query(self, query: str, params: dict = None, timeout: int = None):
         query = str(query)
         timeout = self._timeout if timeout is None else timeout
         # using the query API to get raw data
-        response = self._session.get(
-            "{0}/api/v1/query".format(self.url),
+        response = self._session.request(
+            method=self._method,
+            url="{0}/api/v1/query".format(self.url),
             params={**{"query": query}, **params},
             verify=self._session.verify,
             headers=self.headers,
@@ -490,8 +508,9 @@ def custom_query_range(
         query = str(query)
         timeout = self._timeout if timeout is None else timeout
         # using the query_range API to get raw data
-        response = self._session.get(
-            "{0}/api/v1/query_range".format(self.url),
+        response = self._session.request(
+            method=self._method,
+            url="{0}/api/v1/query_range".format(self.url),
             params={**{"query": query, "start": start, "end": end, "step": step}, **params},
             verify=self._session.verify,
             headers=self.headers,
@@ -626,8 +645,9 @@ def get_targets(self, state: str = None, scrape_pool: str = None):
         if scrape_pool:
             params['scrapePool'] = scrape_pool
 
-        response = self._session.get(
-            "{0}/api/v1/targets".format(self.url),
+        response = self._session.request(
+            method="GET",
+            url="{0}/api/v1/targets".format(self.url),
             verify=self._session.verify,
             headers=self.headers,
             params=params,
@@ -666,8 +686,9 @@ def get_target_metadata(self, target: dict[str, str], metric: str = None):
                 ",".join(f'{k}="{v}"' for k, v in target.items()) + "}"
             params['match_target'] = match_target
 
-        response = self._session.get(
-            "{0}/api/v1/targets/metadata".format(self.url),
+        response = self._session.request(
+            method="GET",
+            url="{0}/api/v1/targets/metadata".format(self.url),
             verify=self._session.verify,
             headers=self.headers,
             params=params,
@@ -708,8 +729,9 @@ def get_metric_metadata(self, metric: str, limit: int = None, limit_per_metric:
         if limit_per_metric:
             params['limit_per_metric'] = limit_per_metric
 
-        response = self._session.get(
-            "{0}/api/v1/metadata".format(self.url),
+        response = self._session.request(
+            method="GET",
+            url="{0}/api/v1/metadata".format(self.url),
             verify=self._session.verify,
             headers=self.headers,
             params=params,

tests/test_prometheus_connect.py

@@ -219,6 +219,65 @@ def test_get_metric_metadata(self):  # PR #295
         limited_per_metric = self.pc.get_metric_metadata(metric_name, limit_per_metric=1)
         self.assertIsInstance(limited_per_metric, list)
 
+    def test_method_argument_accepts_get_and_post(self):
+        """Test that PrometheusConnect accepts GET and POST for method argument, and raises on invalid values."""
+        # Default should be GET
+        pc_default = PrometheusConnect(url=self.prometheus_host, disable_ssl=False)
+        self.assertEqual(pc_default._method, "GET")
+        # Explicit GET
+        pc_get = PrometheusConnect(url=self.prometheus_host, disable_ssl=False, method="GET")
+        self.assertEqual(pc_get._method, "GET")
+        # Explicit POST
+        pc_post = PrometheusConnect(url=self.prometheus_host, disable_ssl=False, method="POST")
+        self.assertEqual(pc_post._method, "POST")
+        # Invalid type
+        with self.assertRaises(TypeError):
+            PrometheusConnect(url=self.prometheus_host, disable_ssl=False, method=123)
+        # Invalid value
+        with self.assertRaises(ValueError):
+            PrometheusConnect(url=self.prometheus_host, disable_ssl=False, method="PUT")
+            
+    def test_post_method_for_supported_functions(self):
+        """Test that PrometheusConnect uses POST for supported endpoints when method='POST', and returns a value."""
+        pc = PrometheusConnect(url=self.prometheus_host, disable_ssl=False, method="POST")
+        start_time = datetime.now() - timedelta(minutes=10)
+        end_time = datetime.now()
+
+        # custom_query should use POST and return something (or raise)
+        try:
+            result = pc.custom_query("up")
+            self.assertTrue(result is not None and result != [], "no metrics received from prometheus")
+        except Exception as e:
+            self.fail(f"custom_query('up') raised an unexpected exception: {e}")
+
+        # custom_query_range should use POST and return something (or raise)
+        try:
+            result = pc.custom_query_range("up", start_time=start_time, end_time=end_time, step="15")
+            self.assertTrue(result is not None and result != [], "no metrics received from prometheus")
+        except Exception as e:
+            self.fail(f"custom_query_range('up', ...) raised an unexpected exception: {e}")
+
+        # get_label_names should use POST and return something (or raise)
+        try:
+            result = pc.get_label_names()
+            self.assertTrue(result is not None and result != [], "no metrics received from prometheus")
+        except Exception as e:
+            self.fail(f"get_label_names() raised an unexpected exception: {e}")
+
+        # get_current_metric_value should use POST and return something (or raise)
+        try:
+            result = pc.get_current_metric_value("up")
+            self.assertTrue(result is not None and result != [], "no metrics received from prometheus")
+        except Exception as e:
+            self.fail(f"get_current_metric_value('up') raised an unexpected exception: {e}")
+
+        # get_metric_range_data should use POST and return something (or raise)
+        try:
+            result = pc.get_metric_range_data("up", start_time=start_time, end_time=end_time)
+            self.assertTrue(result is not None and result != [], "no metrics received from prometheus")
+        except Exception as e:
+            self.fail(f"get_metric_range_data('up', ...) raised an unexpected exception: {e}")
+
 
 class TestPrometheusConnectWithMockedNetwork(BaseMockedNetworkTestcase):
     """Network is blocked in this testcase, see base class."""