Wavyzz/py-healthchecks.io-fork
1
0
Fork 0
forked from Wavyzz/py-healthchecks.io
Code Pull Requests Activity

ping api and tests

Browse Source
This commit is contained in:
Andrew Herrington
2021-12-11 23:38:03 -06:00
parent 7f1ca2d331
commit fb50209a5f
12 changed files with 719 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
src/healthchecks_io/client/sync_client.py
View File

@@ -2,6 +2,7 @@
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
from httpx import Client as HTTPXClient
@@ -17,22 +18,32 @@ class Client(AbstractClient):
def __init__(
self,
api_key: str,
api_key: str = "",
ping_key: str = "",
api_url: str = "https://healthchecks.io/api/",
ping_url: str = "https://hc-ping.com/",
api_version: int = 1,
client: Optional[HTTPXClient] = None,
) -> None:
"""An AsyncClient can be used in code using asyncio to work with the Healthchecks.io api.
Args:
api_key (str): Healthchecks.io API key
api_key (str): Healthchecks.io API key. Defaults to an empty string.
ping_key (str): Healthchecks.io Ping key. Defaults to an empty string.
api_url (str): API URL. Defaults to "https://healthchecks.io/api/".
ping_url (str): Ping API url. Defaults to "https://hc-ping.com/"
api_version (int): Versiopn of the api to use. Defaults to 1.
client (Optional[HTTPXClient], optional): A httpx.Client. If not
passed in, one will be created for this object. Defaults to None.
"""
self._client: HTTPXClient = HTTPXClient() if client is None else client
super().__init__(api_key=api_key, api_url=api_url, api_version=api_version)
super().__init__(
api_key=api_key,
ping_key=ping_key,
api_url=api_url,
ping_url=ping_url,
api_version=api_version,
)
self._client.headers["X-Api-Key"] = self._api_key
self._client.headers["user-agent"] = f"py-healthchecks.io/{VERSION}"
self._client.headers["Content-type"] = "application/json"
@@ -247,3 +258,140 @@ class Client(AbstractClient):
key: badges.Badges.from_api_result(item)
for key, item in response.json()["badges"].items()
}
def success_ping(self, uuid: str = "", slug: 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.
Can take a uuid or a slug. If you call with a slug, you much have a
ping key set.
Check's slug is not guaranteed to be unique. If multiple checks in the
project have the same name, they also have the same slug. If you make
a Pinging API request using a non-unique slug, Healthchecks.io will
return the "409 Conflict" HTTP status code and ignore the request.
Args:
uuid (str): Check's UUID. Defaults to "".
slug (str): Check's Slug. Defaults to "".
Raises:
HCAPIAuthError: Raised when status_code == 401 or 403
HCAPIError: Raised when status_code is 5xx
CheckNotFoundError: Raised when status_code is 404 or response text has "not found" in it
BadAPIRequestError: Raised when status_code is 400, or if you pass a uuid and a slug, or if
pinging by a slug and do not have a ping key set
HCAPIRateLimitError: Raised when status code is 429 or response text has "rate limited" in it
NonUniqueSlugError: Raused when status code is 409.
Returns:
Tuple[bool, str]: success (true or false) and the response text
"""
ping_url = self._get_ping_url(uuid, slug, "")
response = self.check_ping_response(self._client.get(ping_url))
return (True if response.status_code == 200 else False, response.text)
def start_ping(self, uuid: str = "", slug: 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:
* Healthchecks.io will measure and display job execution times
* Healthchecks.io will detect if the job runs longer than its configured grace time
Can take a uuid or a slug. If you call with a slug, you much have a
ping key set.
Check's slug is not guaranteed to be unique. If multiple checks in the
project have the same name, they also have the same slug. If you make
a Pinging API request using a non-unique slug, Healthchecks.io will
return the "409 Conflict" HTTP status code and ignore the request.
Args:
uuid (str): Check's UUID. Defaults to "".
slug (str): Check's Slug. Defaults to "".
Raises:
HCAPIAuthError: Raised when status_code == 401 or 403
HCAPIError: Raised when status_code is 5xx
CheckNotFoundError: Raised when status_code is 404 or response text has "not found" in it
BadAPIRequestError: Raised when status_code is 400, or if you pass a uuid and a slug, or if
pinging by a slug and do not have a ping key set
HCAPIRateLimitError: Raised when status code is 429 or response text has "rate limited" in it
NonUniqueSlugError: Raused when status code is 409.
Returns:
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(self._client.get(ping_url))
return (True if response.status_code == 200 else False, response.text)
def fail_ping(self, uuid: str = "", slug: 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.
Can take a uuid or a slug. If you call with a slug, you much have a
ping key set.
Check's slug is not guaranteed to be unique. If multiple checks in the
project have the same name, they also have the same slug. If you make
a Pinging API request using a non-unique slug, Healthchecks.io will
return the "409 Conflict" HTTP status code and ignore the request.
Args:
uuid (str): Check's UUID. Defaults to "".
slug (str): Check's Slug. Defaults to "".
Raises:
HCAPIAuthError: Raised when status_code == 401 or 403
HCAPIError: Raised when status_code is 5xx
CheckNotFoundError: Raised when status_code is 404 or response text has "not found" in it
BadAPIRequestError: Raised when status_code is 400, or if you pass a uuid and a slug, or if
pinging by a slug and do not have a ping key set
HCAPIRateLimitError: Raised when status code is 429 or response text has "rate limited" in it
NonUniqueSlugError: Raused when status code is 409.
Returns:
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(self._client.get(ping_url))
return (True if response.status_code == 200 else False, response.text)
def exit_code_ping(
self, exit_code: int, uuid: str = "", slug: 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.
Can take a uuid or a slug. If you call with a slug, you much have a
ping key set.
Check's slug is not guaranteed to be unique. If multiple checks in the
project have the same name, they also have the same slug. If you make
a Pinging API request using a non-unique slug, Healthchecks.io will
return the "409 Conflict" HTTP status code and ignore the request.
Args:
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 "".
Raises:
HCAPIAuthError: Raised when status_code == 401 or 403
HCAPIError: Raised when status_code is 5xx
CheckNotFoundError: Raised when status_code is 404 or response text has "not found" in it
BadAPIRequestError: Raised when status_code is 400, or if you pass a uuid and a slug, or if
pinging by a slug and do not have a ping key set
HCAPIRateLimitError: Raised when status code is 429 or response text has "rate limited" in it
NonUniqueSlugError: Raused when status code is 409.
Returns:
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(self._client.get(ping_url))
return (True if response.status_code == 200 else False, response.text)