Wavyzz/py-healthchecks.io
1
0
Fork 1
mirror of https://github.com/andrewthetechie/py-healthchecks.io.git synced 2026-05-01 11:17:06 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: add python 3.11 support (#392)

Browse Source 
* feat: add python 3.11 support

- adds testing for python 3.11
- moves from black and flake8 to ruff for linting
- Updates coverage output from tests

* fix: fix dev dependency group

* fix: add installing dev dependencies

* ci: add toml

* ci: simplify coverage

* ci: update coverage
This commit is contained in:
Andrew
2023-02-20 23:12:01 -06:00
committed by GitHub
parent 6ae5ecefea
commit a752c42a28
15 changed files with 198 additions and 531 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/healthchecks_io/client/_abstract.py
View File

@@ -83,7 +83,8 @@ class AbstractClient(ABC):
endpoint (str): Endpoint to request
Raises:
BadAPIRequestError: Raised if you pass a uuid and a slug, or if pinging by a slug and do not have a ping key set
BadAPIRequestError: Raised if you pass a uuid and a slug, or if pinging by a slug and do not have a
ping key set
Returns:
str: url for this

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

@@ -42,9 +42,7 @@ class AsyncClient(AbstractClient):
client (Optional[HTTPXAsyncClient], optional): A httpx.Asyncclient. If not
passed in, one will be created for this object. Defaults to None.
"""
self._client: HTTPXAsyncClient = (
HTTPXAsyncClient() if client is None else client
)
self._client: HTTPXAsyncClient = HTTPXAsyncClient() if client is None else client
super().__init__(
api_key=api_key,
ping_key=ping_key,
@@ -53,9 +51,7 @@ class AsyncClient(AbstractClient):
api_version=api_version,
)
self._client.headers["X-Api-Key"] = self._api_key
self._client.headers[
"user-agent"
] = f"py-healthchecks.io-async/{client_version}"
self._client.headers["user-agent"] = f"py-healthchecks.io-async/{client_version}"
self._client.headers["Content-type"] = "application/json"
async def __aenter__(self) -> "AsyncClient":
@@ -97,9 +93,7 @@ class AsyncClient(AbstractClient):
Check: check that was just created
"""
request_url = self._get_api_request_url("checks/")
response = self.check_response(
await self._client.post(request_url, json=new_check.dict(exclude_none=True))
)
response = self.check_response(await self._client.post(request_url, json=new_check.dict(exclude_none=True)))
return Check.from_api_result(response.json())
async def update_check(self, uuid: str, update_check: CheckCreate) -> Check:
@@ -143,16 +137,11 @@ class AsyncClient(AbstractClient):
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
)
request_url = self._add_url_params(request_url, {"tag": tag}, replace=False)
response = self.check_response(await self._client.get(request_url))
return [
Check.from_api_result(check_data)
for check_data in response.json()["checks"]
]
return [Check.from_api_result(check_data) for check_data in response.json()["checks"]]
async def get_check(self, check_id: str) -> Check:
"""Get a single check by id.
@@ -247,10 +236,7 @@ class AsyncClient(AbstractClient):
"""
request_url = self._get_api_request_url(f"checks/{check_id}/pings/")
response = self.check_response(await self._client.get(request_url))
return [
CheckPings.from_api_result(check_data)
for check_data in response.json()["pings"]
]
return [CheckPings.from_api_result(check_data) for check_data in response.json()["pings"]]
async def get_check_flips(
self,
@@ -274,8 +260,10 @@ class AsyncClient(AbstractClient):
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.
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[CheckStatuses]: List of status flips for this check
@@ -307,10 +295,7 @@ class AsyncClient(AbstractClient):
"""
request_url = self._get_api_request_url("channels/")
response = self.check_response(await self._client.get(request_url))
return [
Integration.from_api_result(integration_dict)
for integration_dict in response.json()["channels"]
]
return [Integration.from_api_result(integration_dict) for integration_dict in response.json()["channels"]]
async def get_badges(self) -> Dict[str, Badges]:
"""Returns a dict of all tags in the project, with badge URLs for each tag.
@@ -321,7 +306,8 @@ class AsyncClient(AbstractClient):
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".
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
@@ -337,14 +323,9 @@ class AsyncClient(AbstractClient):
"""
request_url = self._get_api_request_url("badges/")
response = self.check_response(await self._client.get(request_url))
return {
key: Badges.from_api_result(item)
for key, item in response.json()["badges"].items()
}
return {key: Badges.from_api_result(item) for key, item in response.json()["badges"].items()}
async def success_ping(
self, uuid: str = "", slug: str = "", data: 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.
@@ -375,14 +356,10 @@ 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.post(ping_url, content=data)
)
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 = "", data: 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:
@@ -415,14 +392,10 @@ 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.post(ping_url, content=data)
)
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 = "", data: 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.
@@ -453,14 +426,10 @@ 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.post(ping_url, content=data)
)
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 = "", data: str = ""
) -> Tuple[bool, str]:
async def exit_code_ping(self, exit_code: int, 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.
@@ -492,7 +461,5 @@ 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.post(ping_url, content=data)
)
response = self.check_ping_response(await self._client.post(ping_url, content=data))
return (True if response.status_code == 200 else False, response.text)

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

@@ -90,16 +90,11 @@ class Client(AbstractClient):
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
)
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"]
]
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.
@@ -137,9 +132,7 @@ class Client(AbstractClient):
Check: check that was just created
"""
request_url = self._get_api_request_url("checks/")
response = self.check_response(
self._client.post(request_url, json=new_check.dict(exclude_none=True))
)
response = self.check_response(self._client.post(request_url, json=new_check.dict(exclude_none=True)))
return Check.from_api_result(response.json())
def update_check(self, uuid: str, update_check: CheckCreate) -> Check:
@@ -236,10 +229,7 @@ class Client(AbstractClient):
"""
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"]
]
return [checks.CheckPings.from_api_result(check_data) for check_data in response.json()["pings"]]
def get_check_flips(
self,
@@ -261,8 +251,10 @@ class Client(AbstractClient):
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.
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
@@ -307,7 +299,8 @@ class Client(AbstractClient):
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".
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
@@ -322,14 +315,9 @@ class Client(AbstractClient):
"""
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()
}
return {key: badges.Badges.from_api_result(item) for key, item in response.json()["badges"].items()}
def success_ping(
self, uuid: str = "", slug: str = "", data: str = ""
) -> Tuple[bool, str]:
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.
@@ -363,9 +351,7 @@ class Client(AbstractClient):
response = self.check_ping_response(self._client.post(ping_url, content=data))
return (True if response.status_code == 200 else False, response.text)
def start_ping(
self, uuid: str = "", slug: str = "", data: str = ""
) -> Tuple[bool, str]:
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:
@@ -401,9 +387,7 @@ class Client(AbstractClient):
response = self.check_ping_response(self._client.post(ping_url, content=data))
return (True if response.status_code == 200 else False, response.text)
def fail_ping(
self, uuid: str = "", slug: str = "", data: str = ""
) -> Tuple[bool, str]:
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.
@@ -437,9 +421,7 @@ class Client(AbstractClient):
response = self.check_ping_response(self._client.post(ping_url, content=data))
return (True if response.status_code == 200 else False, response.text)
def exit_code_ping(
self, exit_code: int, uuid: str = "", slug: str = "", data: str = ""
) -> Tuple[bool, str]:
def exit_code_ping(self, exit_code: int, 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.

23
src/healthchecks_io/schemas/checks.py
View File

@@ -42,15 +42,14 @@ class Check(BaseModel):
uuid: Optional[str]
@validator("uuid", always=True)
def validate_uuid(
cls, value: Optional[str], values: Dict[str, Any] # noqa: B902
) -> Optional[str]:
def validate_uuid(cls, value: Optional[str], values: Dict[str, Any]) -> Optional[str]: # noqa: B902
"""Tries to set the uuid from the ping_url.
Will return none if a read only token is used because it cannot retrieve the UUID of a check
"""
if value is None and values.get("ping_url", None) is not None:
# url is like healthchecks.io/ping/8f57b84b-86c2-4546-8923-03f83d27604a, so we want just the UUID off the end
# url is like healthchecks.io/ping/8f57b84b-86c2-4546-8923-03f83d27604a, so we want just the
# UUID off the end
# Parse the url, grab the path and then just get the name using pathlib
path = PurePath(str(urlparse(values.get("ping_url")).path))
return path.name
@@ -66,9 +65,7 @@ class CheckCreate(BaseModel):
"""Pydantic object for creating a check."""
name: Optional[str] = Field("", description="Name of the check")
tags: Optional[str] = Field(
"", description="String separated list of tags to apply"
)
tags: Optional[str] = Field("", description="String separated list of tags to apply")
desc: Optional[str] = Field("", description="Description of the check")
timeout: Optional[int] = Field(
86400,
@@ -147,9 +144,7 @@ class CheckCreate(BaseModel):
def validate_methods(cls, value: str) -> str:
"""Validate that methods."""
if value not in ("", "POST"):
raise ValueError(
"Methods is invalid, it should be either an empty string or POST"
)
raise ValueError("Methods is invalid, it should be either an empty string or POST")
return value
@validator("unique")
@@ -167,9 +162,7 @@ class CheckUpdate(CheckCreate):
"""Pydantic object for updating a check."""
name: Optional[str] = Field(None, description="Name of the check")
tags: Optional[str] = Field(
None, description="String separated list of tags to apply"
)
tags: Optional[str] = Field(None, description="String separated list of tags to apply")
timeout: Optional[int] = Field(
None,
description="The expected period of this check in seconds.",
@@ -243,9 +236,7 @@ class CheckPings(BaseModel):
duration: Optional[float] = None
@classmethod
def from_api_result(
cls, ping_dict: Dict[str, Union[str, int, datetime]]
) -> "CheckPings":
def from_api_result(cls, ping_dict: Dict[str, Union[str, int, datetime]]) -> "CheckPings":
"""Converts a dictionary from the healthchecks api into a CheckPings object."""
ping_dict["number_of_pings"] = ping_dict["n"]
ping_dict["user_agent"] = ping_dict["ua"]