diff --git a/src/healthchecks_io/__init__.py b/src/healthchecks_io/__init__.py index a9d1761..89ec522 100644 --- a/src/healthchecks_io/__init__.py +++ b/src/healthchecks_io/__init__.py @@ -1,3 +1,9 @@ """Py Healthchecks.Io.""" +VERSION = "0.1" # noqa: E402 -VERSION = "0.1" +from .client import AsyncClient # noqa: F401, E402 +from .client import Client # noqa: F401, E402 +from .client.exceptions import BadAPIRequestError # noqa: F401, E402 +from .client.exceptions import CheckNotFoundError # noqa: F401, E402 +from .client.exceptions import HCAPIAuthError # noqa: F401, E402 +from .client.exceptions import HCAPIError # noqa: F401, E402 diff --git a/src/healthchecks_io/client/__init__.py b/src/healthchecks_io/client/__init__.py index 8a97a7e..2c40e4c 100644 --- a/src/healthchecks_io/client/__init__.py +++ b/src/healthchecks_io/client/__init__.py @@ -1,2 +1,3 @@ """healthchecks_io clients.""" -from .asyncclient import AsyncClient # noqa: F401 +from .async_client import AsyncClient # noqa: F401 +from .sync_client import Client # noqa: F401 diff --git a/src/healthchecks_io/client/asyncclient.py b/src/healthchecks_io/client/async_client.py similarity index 99% rename from src/healthchecks_io/client/asyncclient.py rename to src/healthchecks_io/client/async_client.py index c33c886..075f701 100644 --- a/src/healthchecks_io/client/asyncclient.py +++ b/src/healthchecks_io/client/async_client.py @@ -38,7 +38,7 @@ class AsyncClient(AbstractClient): api_key=api_key, api_url=api_url, api_version=api_version, client=client ) self._client.headers["X-Api-Key"] = self._api_key - self._client.headers["user-agent"] = f"py-healthchecks.io/{VERSION}" + self._client.headers["user-agent"] = f"py-healthchecks.io-async/{VERSION}" self._client.headers["Content-type"] = "application/json" def _finalizer_method(self): diff --git a/src/healthchecks_io/client/sync_client.py b/src/healthchecks_io/client/sync_client.py new file mode 100644 index 0000000..3f44998 --- /dev/null +++ b/src/healthchecks_io/client/sync_client.py @@ -0,0 +1,252 @@ +"""An async healthchecks.io client.""" +from typing import Dict +from typing import List +from typing import Optional + +from httpx import Client as HTTPXClient + +from ._abstract import AbstractClient +from healthchecks_io import VERSION +from healthchecks_io.schemas import badges +from healthchecks_io.schemas import checks +from healthchecks_io.schemas import integrations + + +class Client(AbstractClient): + """A Healthchecks.io client implemented using httpx's sync methods.""" + + def __init__( + self, + api_key: str, + api_url: Optional[str] = "https://healthchecks.io/api/", + api_version: Optional[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_url (Optional[str], optional): API URL. Defaults to "https://healthchecks.io/api/". + api_version (Optional[int], optional): 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. + """ + if client is None: + client = HTTPXClient() + super().__init__( + api_key=api_key, api_url=api_url, api_version=api_version, client=client + ) + 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" + + def _finalizer_method(self): + """Closes the httpx client.""" + self._client.close() + + def get_checks(self, tags: Optional[List[str]] = None) -> List[checks.Check]: + """Get a list of checks from the healthchecks api. + + Args: + tags (Optional[List[str]], optional): Filters the checks and returns only + the checks that are tagged with the specified value. Defaults to None. + + Raises: + HCAPIAuthError: When the API returns a 401, indicates an api key issue + HCAPIError: When the API returns anything other than a 200 or 401 + + Returns: + List[checks.Check]: [description] + """ + request_url = self._get_api_request_url("checks/") + if tags is not None: + for tag in tags: + request_url = self._add_url_params( + request_url, {"tag": tag}, replace=False + ) + + response = self.check_response(self._client.get(request_url)) + + return [ + checks.Check.from_api_result(check_data) + for check_data in response.json()["checks"] + ] + + def get_check(self, check_id: str) -> checks.Check: + """Get a single check by id. + + check_id can either be a check uuid if using a read/write api key + or a unique key if using a read only api key. + + Args: + check_id (str): check's uuid or unique id + + Returns: + checks.Check: the check + + Raises: + HCAPIAuthError: Raised when status_code == 401 or 403 + HCAPIError: Raised when status_code is 5xx + CheckNotFoundError: Raised when status_code is 404 + + """ + request_url = self._get_api_request_url(f"checks/{check_id}") + response = self.check_response(self._client.get(request_url)) + return checks.Check.from_api_result(response.json()) + + def pause_check(self, check_id: str) -> checks.Check: + """Disables monitoring for a check without removing it. + + The check goes into a "paused" state. + You can resume monitoring of the check by pinging it. + + check_id must be a uuid, not a unique id + + Args: + check_id (str): check's uuid + + Returns: + checks.Check: the check just paused + + Raises: + HCAPIAuthError: Raised when status_code == 401 or 403 + HCAPIError: Raised when status_code is 5xx + CheckNotFoundError: Raised when status_code is 404 + + """ + request_url = self._get_api_request_url(f"checks/{check_id}/pause") + response = self.check_response(self._client.post(request_url, data={})) + return checks.Check.from_api_result(response.json()) + + def delete_check(self, check_id: str) -> checks.Check: + """Permanently deletes the check from the user's account. + + check_id must be a uuid, not a unique id + + Args: + check_id (str): check's uuid + + Returns: + checks.Check: the check just deleted + + Raises: + HCAPIAuthError: Raised when status_code == 401 or 403 + HCAPIError: Raised when status_code is 5xx + CheckNotFoundError: Raised when status_code is 404 + + """ + request_url = self._get_api_request_url(f"checks/{check_id}") + response = self.check_response(self._client.delete(request_url)) + return checks.Check.from_api_result(response.json()) + + def get_check_pings(self, check_id: str) -> List[checks.CheckPings]: + """Returns a list of pings this check has received. + + This endpoint returns pings in reverse order (most recent first), + and the total number of returned pings depends on the account's + billing plan: 100 for free accounts, 1000 for paid accounts. + + Args: + check_id (str): check's uuid + + Returns: + List[checks.CheckPings]: list of pings this check has received + + Raises: + HCAPIAuthError: Raised when status_code == 401 or 403 + HCAPIError: Raised when status_code is 5xx + CheckNotFoundError: Raised when status_code is 404 + + """ + request_url = self._get_api_request_url(f"checks/{check_id}/pings/") + response = self.check_response(self._client.get(request_url)) + return [ + checks.CheckPings.from_api_result(check_data) + for check_data in response.json()["pings"] + ] + + def get_check_flips( + self, + check_id: str, + seconds: Optional[int] = None, + start: Optional[int] = None, + end: Optional[int] = None, + ) -> List[checks.CheckStatuses]: + """Returns a list of "flips" this check has experienced. + + A flip is a change of status (from "down" to "up," or from "up" to "down"). + + Raises: + HCAPIAuthError: Raised when status_code == 401 or 403 + HCAPIError: Raised when status_code is 5xx + CheckNotFoundError: Raised when status_code is 404 + BadAPIRequestError: Raised when status_code is 400 + + Args: + check_id (str): check uuid + seconds (Optional[int], optional): Returns the flips from the last value seconds. Defaults to None. + start (Optional[int], optional): Returns flips that are newer than the specified UNIX timestamp.. Defaults to None. + end (Optional[int], optional): Returns flips that are older than the specified UNIX timestamp.. Defaults to None. + + Returns: + List[checks.CheckStatuses]: List of status flips for this check + + """ + params = dict() + if seconds is not None and seconds >= 0: + params["seconds"] = seconds + if start is not None and start >= 0: + params["start"] = start + if end is not None and end >= 0: + params["end"] = end + + request_url = self._get_api_request_url(f"checks/{check_id}/flips/", params) + response = self.check_response(self._client.get(request_url)) + return [checks.CheckStatuses(**status_data) for status_data in response.json()] + + def get_integrations(self) -> List[Optional[integrations.Integration]]: + """Returns a list of integrations belonging to the project. + + Raises: + HCAPIAuthError: Raised when status_code == 401 or 403 + HCAPIError: Raised when status_code is 5xx + + Returns: + List[Optional[integrations.Integration]]: List of integrations for the project + + """ + request_url = self._get_api_request_url("channels/") + response = self.check_response(self._client.get(request_url)) + return [ + integrations.Integration.from_api_result(integration_dict) + for integration_dict in response.json()["channels"] + ] + + def get_badges(self) -> Dict[str, badges.Badges]: + """Returns a dict of all tags in the project, with badge URLs for each tag. + + Healthchecks.io provides badges in a few different formats: + svg: returns the badge as a SVG document. + json: returns a JSON document which you can use to generate a custom badge yourself. + shields: returns JSON in a Shields.io compatible format. + In addition, badges have 2-state and 3-state variations: + + svg, json, shields: reports two states: "up" and "down". It considers any checks in the grace period as still "up". + svg3, json3, shields3: reports three states: "up", "late", and "down". + + The response includes a special * entry: this pseudo-tag reports the overal status + of all checks in the project. + + Raises: + HCAPIAuthError: Raised when status_code == 401 or 403 + HCAPIError: Raised when status_code is 5xx + + Returns: + Dict[str, badges.Badges]: Dictionary of all tags in the project with badges + """ + request_url = self._get_api_request_url("badges/") + response = self.check_response(self._client.get(request_url)) + return { + key: badges.Badges.from_api_result(item) + for key, item in response.json()["badges"].items() + } diff --git a/tests/client/test_sync.py b/tests/client/test_sync.py new file mode 100644 index 0000000..55fb863 --- /dev/null +++ b/tests/client/test_sync.py @@ -0,0 +1,197 @@ +from urllib.parse import urljoin + +import pytest +import respx +from httpx import Client as HTTPXClient +from httpx import Response + +from healthchecks_io import Client +from healthchecks_io.client.exceptions import BadAPIRequestError +from healthchecks_io.client.exceptions import CheckNotFoundError +from healthchecks_io.client.exceptions import HCAPIAuthError +from healthchecks_io.client.exceptions import HCAPIError + + +@pytest.mark.respx +def test_get_checks_200(fake_check_api_result, respx_mock, test_client): + assert test_client._client is not None + checks_url = urljoin(test_client._api_url, "checks/") + respx_mock.get(checks_url).mock( + return_value=Response(status_code=200, json={"checks": [fake_check_api_result]}) + ) + checks = test_client.get_checks() + assert len(checks) == 1 + assert checks[0].name == fake_check_api_result["name"] + + +@pytest.mark.respx +def test_get_checks_pass_in_client(fake_check_api_result, respx_mock): + httpx_client = HTTPXClient() + test_client = Client( + api_key="test", api_url="http://localhost/api/", client=httpx_client + ) + checks_url = urljoin(test_client._api_url, "checks/") + respx_mock.get(checks_url).mock( + return_value=Response(status_code=200, json={"checks": [fake_check_api_result]}) + ) + checks = test_client.get_checks() + assert len(checks) == 1 + assert checks[0].name == fake_check_api_result["name"] + + +@pytest.mark.respx +def test_get_checks_exceptions(fake_check_api_result, respx_mock, test_client): + checks_url = urljoin(test_client._api_url, "checks/") + # test exceptions + respx_mock.get(checks_url).mock(return_value=Response(status_code=401)) + with pytest.raises(HCAPIAuthError): + test_client.get_checks() + + respx_mock.get(checks_url).mock(return_value=Response(status_code=500)) + with pytest.raises(HCAPIError): + test_client.get_checks() + + +@pytest.mark.respx +def test_get_checks_tags(fake_check_api_result, respx_mock, test_client): + """Test get_checks with tags""" + checks_url = urljoin(test_client._api_url, "checks/") + respx_mock.get(f"{checks_url}?tag=test&tag=test2").mock( + return_value=Response(status_code=200, json={"checks": [fake_check_api_result]}) + ) + checks = test_client.get_checks(tags=["test", "test2"]) + assert len(checks) == 1 + assert checks[0].name == fake_check_api_result["name"] + + +def test_finalizer_closes(test_client): + """Tests our finalizer works to close the method""" + assert not test_client.is_closed + test_client._finalizer_method() + assert test_client.is_closed + + +@pytest.mark.respx +def test_get_check_200(fake_check_api_result, respx_mock, test_client): + assert test_client._client is not None + checks_url = urljoin(test_client._api_url, "checks/test") + respx_mock.get(checks_url).mock( + return_value=Response(status_code=200, json=fake_check_api_result) + ) + check = test_client.get_check(check_id="test") + assert check.name == fake_check_api_result["name"] + + +@pytest.mark.respx +def test_check_get_404(respx_mock, test_client): + assert test_client._client is not None + checks_url = urljoin(test_client._api_url, "checks/test") + respx_mock.get(checks_url).mock(return_value=Response(status_code=404)) + with pytest.raises(CheckNotFoundError): + test_client.get_check("test") + + +@pytest.mark.respx +def test_pause_check_200(fake_check_api_result, respx_mock, test_client): + checks_url = urljoin(test_client._api_url, "checks/test/pause") + respx_mock.post(checks_url).mock( + return_value=Response(status_code=200, json=fake_check_api_result) + ) + check = test_client.pause_check(check_id="test") + assert check.name == fake_check_api_result["name"] + + +@pytest.mark.respx +def test_check_pause_404(respx_mock, test_client): + assert test_client._client is not None + checks_url = urljoin(test_client._api_url, "checks/test/pause") + respx_mock.post(checks_url).mock(return_value=Response(status_code=404)) + with pytest.raises(CheckNotFoundError): + test_client.pause_check("test") + + +@pytest.mark.respx +def test_delete_check_200(fake_check_api_result, respx_mock, test_client): + assert test_client._client is not None + checks_url = urljoin(test_client._api_url, "checks/test") + respx_mock.delete(checks_url).mock( + return_value=Response(status_code=200, json=fake_check_api_result) + ) + check = test_client.delete_check(check_id="test") + assert check.name == fake_check_api_result["name"] + + +@pytest.mark.respx +def test_delete_pause404(respx_mock, test_client): + checks_url = urljoin(test_client._api_url, "checks/test") + respx_mock.delete(checks_url).mock(return_value=Response(status_code=404)) + with pytest.raises(CheckNotFoundError): + test_client.delete_check("test") + + +@pytest.mark.respx +def test_get_check_pings_200(fake_check_pings_api_result, respx_mock, test_client): + checks_url = urljoin(test_client._api_url, "checks/test/pings/") + respx_mock.get(checks_url).mock( + return_value=Response( + status_code=200, json={"pings": fake_check_pings_api_result} + ) + ) + pings = test_client.get_check_pings("test") + assert len(pings) == len(fake_check_pings_api_result) + assert pings[0].type == fake_check_pings_api_result[0]["type"] + + +@pytest.mark.respx +def test_get_check_flips_200(fake_check_flips_api_result, respx_mock, test_client): + checks_url = urljoin(test_client._api_url, "checks/test/flips/") + respx_mock.get(checks_url).mock( + return_value=Response(status_code=200, json=fake_check_flips_api_result) + ) + flips = test_client.get_check_flips("test") + assert len(flips) == len(fake_check_flips_api_result) + assert flips[0].up == fake_check_flips_api_result[0]["up"] + + +@pytest.mark.respx +def test_get_check_flips_params_200( + fake_check_flips_api_result, respx_mock, test_client +): + checks_url = urljoin( + test_client._api_url, "checks/test/flips/?seconds=1&start=1&end=1" + ) + respx_mock.get(checks_url).mock( + return_value=Response(status_code=200, json=fake_check_flips_api_result) + ) + flips = test_client.get_check_flips("test", seconds=1, start=1, end=1) + assert len(flips) == len(fake_check_flips_api_result) + assert flips[0].up == fake_check_flips_api_result[0]["up"] + + +@pytest.mark.respx +def test_get_check_flips_400(fake_check_flips_api_result, respx_mock, test_client): + flips_url = urljoin(test_client._api_url, "checks/test/flips/") + respx_mock.get(flips_url).mock(return_value=Response(status_code=400)) + with pytest.raises(BadAPIRequestError): + test_client.get_check_flips("test") + + +@pytest.mark.respx +def test_get_integrations(fake_integrations_api_result, respx_mock, test_client): + channels_url = urljoin(test_client._api_url, "channels/") + respx_mock.get(channels_url).mock( + return_value=Response(status_code=200, json=fake_integrations_api_result) + ) + integrations = test_client.get_integrations() + assert len(integrations) == len(fake_integrations_api_result["channels"]) + assert integrations[0].id == fake_integrations_api_result["channels"][0]["id"] + + +@pytest.mark.respx +def test_get_badges(fake_badges_api_result, respx_mock, test_client): + channels_url = urljoin(test_client._api_url, "badges/") + respx_mock.get(channels_url).mock( + return_value=Response(status_code=200, json=fake_badges_api_result) + ) + integrations = test_client.get_badges() + assert integrations.keys() == fake_badges_api_result["badges"].keys() diff --git a/tests/conftest.py b/tests/conftest.py index db6f739..b6ac9c0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,7 +4,8 @@ from typing import Union import pytest -from healthchecks_io.client import AsyncClient +from healthchecks_io import AsyncClient +from healthchecks_io import Client from healthchecks_io.schemas import checks @@ -87,6 +88,12 @@ def test_async_client(): yield AsyncClient(api_key="test", api_url="https://localhost/api") +@pytest.fixture +def test_client(): + """A Client for testing, set to a nonsense url so we aren't pinging healtchecks.""" + yield Client(api_key="test", api_url="https://localhost/api") + + @pytest.fixture def fake_check_pings_api_result(): yield [