Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
22344c6
Add HTTP Method as a Parameter for APIs that support POST
Jul 4, 2025
1400375
Update class docstring
Jul 4, 2025
62c6cbc
Add test cases for the method parameter
Jul 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 42 additions & 20 deletions prometheus_api_client/prometheus_connect.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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__(
Expand All @@ -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:
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -129,8 +142,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,
Expand Down Expand Up @@ -160,8 +174,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,
Expand Down Expand Up @@ -212,8 +227,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,
Expand Down Expand Up @@ -299,8 +315,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" + "]",
Expand Down Expand Up @@ -407,8 +424,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,
Expand Down Expand Up @@ -454,8 +472,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,
Expand Down Expand Up @@ -590,8 +609,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,
Expand Down Expand Up @@ -630,8 +650,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,
Expand Down Expand Up @@ -672,8 +693,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,
Expand Down
59 changes: 59 additions & 0 deletions tests/test_prometheus_connect.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,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."""
Expand Down