add badges and integrations

This commit is contained in:
Andrew Herrington
2021-12-10 23:01:00 -06:00
parent b005e77f2b
commit f51effbad7
2 changed files with 93 additions and 0 deletions

View File

@@ -19,7 +19,9 @@ from .exceptions import BadAPIRequestError
from .exceptions import CheckNotFoundError from .exceptions import CheckNotFoundError
from .exceptions import HCAPIAuthError from .exceptions import HCAPIAuthError
from .exceptions import HCAPIError from .exceptions import HCAPIError
from healthchecks_io.schemas import badges
from healthchecks_io.schemas import checks from healthchecks_io.schemas import checks
from healthchecks_io.schemas import integrations
class AbstractClient(ABC): class AbstractClient(ABC):
@@ -177,6 +179,47 @@ class AbstractClient(ABC):
""" """
pass pass
@abstractmethod
def get_integrations(
self,
) -> List[Optional[integrations.Integration]]: # pragma: no cover
"""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
"""
pass
@abstractmethod
def get_badges(self) -> Dict[str, badges.Badges]: # pragma: no cover
"""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
"""
pass
def _get_api_request_url( def _get_api_request_url(
self, path: str, params: Optional[Dict[str, str]] = None self, path: str, params: Optional[Dict[str, str]] = None
) -> str: ) -> str:

View File

@@ -1,5 +1,6 @@
"""An async healthchecks.io client.""" """An async healthchecks.io client."""
import asyncio import asyncio
from typing import Dict
from typing import List from typing import List
from typing import Optional from typing import Optional
@@ -7,7 +8,9 @@ from httpx import AsyncClient as HTTPXAsyncClient
from ._abstract import AbstractClient from ._abstract import AbstractClient
from healthchecks_io import VERSION from healthchecks_io import VERSION
from healthchecks_io.schemas import badges
from healthchecks_io.schemas import checks from healthchecks_io.schemas import checks
from healthchecks_io.schemas import integrations
class AsyncClient(AbstractClient): class AsyncClient(AbstractClient):
@@ -205,3 +208,50 @@ class AsyncClient(AbstractClient):
request_url = self._get_api_request_url(f"checks/{check_id}/flips/", params) request_url = self._get_api_request_url(f"checks/{check_id}/flips/", params)
response = self.check_response(await self._client.get(request_url)) response = self.check_response(await self._client.get(request_url))
return [checks.CheckStatuses(**status_data) for status_data in response.json()] return [checks.CheckStatuses(**status_data) for status_data in response.json()]
async 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(await self._client.get(request_url))
return [
integrations.Integration.from_api_result(integration_dict)
for integration_dict in response.json()["channels"]
]
async 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(await self._client.get(request_url))
return {
key: badges.Badges.from_api_result(item)
for key, item in response.json()["badges"].items()
}