Wavyzz/py-healthchecks.io
1
0
Fork 1
mirror of https://github.com/andrewthetechie/py-healthchecks.io.git synced 2025-12-06 01:28:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

add context manager features

Browse Source
This commit is contained in:
Andrew Herrington
2021-12-30 23:15:08 -06:00
parent 9f139177dc
commit 0606e83f19
10 changed files with 373 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/healthchecks_io/client/async_client.py
View File

@@ -341,7 +341,9 @@ class AsyncClient(AbstractClient):
for key, item in response.json()["badges"].items()
}
async def success_ping(self, uuid: str = "", slug: str = "") -> Tuple[bool, str]:
async def success_ping(
self, uuid: str = "", slug: str = "", data: str = ""
) -> Tuple[bool, str]:
"""Signals to Healthchecks.io that a job has completed successfully.
Can also be used to indicate a continuously running process is still running and healthy.
@@ -357,6 +359,7 @@ class AsyncClient(AbstractClient):
Args:
uuid (str): Check's UUID. Defaults to "".
slug (str): Check's Slug. Defaults to "".
data (str): Text data to append to this check. Defaults to "".
Raises:
HCAPIAuthError: Raised when status_code == 401 or 403
@@ -371,10 +374,14 @@ class AsyncClient(AbstractClient):
Tuple[bool, str]: success (true or false) and the response text
"""
ping_url = self._get_ping_url(uuid, slug, "")
response = self.check_ping_response(await self._client.get(ping_url))
response = self.check_ping_response(
await self._client.post(ping_url, content=data)
)
return (True if response.status_code == 200 else False, response.text)
async def start_ping(self, uuid: str = "", slug: str = "") -> Tuple[bool, str]:
async def start_ping(
self, uuid: str = "", slug: str = "", data: str = ""
) -> Tuple[bool, str]:
"""Sends a "job has started!" message to Healthchecks.io.
Sending a "start" signal is optional, but it enables a few extra features:
@@ -392,6 +399,7 @@ class AsyncClient(AbstractClient):
Args:
uuid (str): Check's UUID. Defaults to "".
slug (str): Check's Slug. Defaults to "".
data (str): Text data to append to this check. Defaults to "".
Raises:
HCAPIAuthError: Raised when status_code == 401 or 403
@@ -406,10 +414,14 @@ class AsyncClient(AbstractClient):
Tuple[bool, str]: success (true or false) and the response text
"""
ping_url = self._get_ping_url(uuid, slug, "/start")
response = self.check_ping_response(await self._client.get(ping_url))
response = self.check_ping_response(
await self._client.post(ping_url, content=data)
)
return (True if response.status_code == 200 else False, response.text)
async def fail_ping(self, uuid: str = "", slug: str = "") -> Tuple[bool, str]:
async def fail_ping(
self, uuid: str = "", slug: str = "", data: str = ""
) -> Tuple[bool, str]:
"""Signals to Healthchecks.io that the job has failed.
Actively signaling a failure minimizes the delay from your monitored service failing to you receiving an alert.
@@ -425,6 +437,7 @@ class AsyncClient(AbstractClient):
Args:
uuid (str): Check's UUID. Defaults to "".
slug (str): Check's Slug. Defaults to "".
data (str): Text data to append to this check. Defaults to "".
Raises:
HCAPIAuthError: Raised when status_code == 401 or 403
@@ -439,11 +452,13 @@ class AsyncClient(AbstractClient):
Tuple[bool, str]: success (true or false) and the response text
"""
ping_url = self._get_ping_url(uuid, slug, "/fail")
response = self.check_ping_response(await self._client.get(ping_url))
response = self.check_ping_response(
await self._client.post(ping_url, content=data)
)
return (True if response.status_code == 200 else False, response.text)
async def exit_code_ping(
self, exit_code: int, uuid: str = "", slug: str = ""
self, exit_code: int, uuid: str = "", slug: str = "", data: str = ""
) -> Tuple[bool, str]:
"""Signals to Healthchecks.io that the job has failed.
@@ -461,6 +476,7 @@ class AsyncClient(AbstractClient):
exit_code (int): Exit code to sent, int from 0 to 255
uuid (str): Check's UUID. Defaults to "".
slug (str): Check's Slug. Defaults to "".
data (str): Text data to append to this check. Defaults to "".
Raises:
HCAPIAuthError: Raised when status_code == 401 or 403
@@ -475,5 +491,7 @@ class AsyncClient(AbstractClient):
Tuple[bool, str]: success (true or false) and the response text
"""
ping_url = self._get_ping_url(uuid, slug, f"/{exit_code}")
response = self.check_ping_response(await self._client.get(ping_url))
response = self.check_ping_response(
await self._client.post(ping_url, content=data)
)
return (True if response.status_code == 200 else False, response.text)