chore: ci fixups and poetry update

This commit is contained in:
Andrew Herrington
2024-05-05 12:20:49 -05:00
parent cb862fa2c5
commit d0b986025e
24 changed files with 829 additions and 1026 deletions

View File

@@ -1,11 +1,7 @@
"""An async healthchecks.io client."""
import asyncio
from types import TracebackType
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
from typing import Type
from httpx import AsyncClient as HTTPXAsyncClient
@@ -29,7 +25,7 @@ class AsyncClient(AbstractClient):
api_url: str = "https://healthchecks.io/api/",
ping_url: str = "https://hc-ping.com/",
api_version: int = 1,
client: Optional[HTTPXAsyncClient] = None,
client: HTTPXAsyncClient | None = None,
) -> None:
"""An AsyncClient can be used in code using asyncio to work with the Healthchecks.io api.
@@ -64,9 +60,9 @@ class AsyncClient(AbstractClient):
async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
exc_type: type[BaseException] | None,
exc: BaseException | None,
traceback: TracebackType | None,
) -> None:
"""Context manager exit."""
await self._afinalizer_method()
@@ -118,7 +114,7 @@ class AsyncClient(AbstractClient):
)
return Check.from_api_result(response.json())
async def get_checks(self, tags: Optional[List[str]] = None) -> List[Check]:
async def get_checks(self, tags: list[str] | None = None) -> list[Check]:
"""Get a list of checks from the healthchecks api.
Args:
@@ -213,7 +209,7 @@ class AsyncClient(AbstractClient):
response = self.check_response(await self._client.delete(request_url))
return Check.from_api_result(response.json())
async def get_check_pings(self, check_id: str) -> List[CheckPings]:
async def get_check_pings(self, check_id: str) -> list[CheckPings]:
"""Returns a list of pings this check has received.
This endpoint returns pings in reverse order (most recent first),
@@ -241,10 +237,10 @@ class AsyncClient(AbstractClient):
async def get_check_flips(
self,
check_id: str,
seconds: Optional[int] = None,
start: Optional[int] = None,
end: Optional[int] = None,
) -> List[CheckStatuses]:
seconds: int | None = None,
start: int | None = None,
end: int | None = None,
) -> list[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").
@@ -281,7 +277,7 @@ class AsyncClient(AbstractClient):
response = self.check_response(await self._client.get(request_url))
return [CheckStatuses(**status_data) for status_data in response.json()]
async def get_integrations(self) -> List[Optional[Integration]]:
async def get_integrations(self) -> list[Integration | None]:
"""Returns a list of integrations belonging to the project.
Raises:
@@ -297,7 +293,7 @@ class AsyncClient(AbstractClient):
response = self.check_response(await self._client.get(request_url))
return [Integration.from_api_result(integration_dict) for integration_dict in response.json()["channels"]]
async def get_badges(self) -> Dict[str, Badges]:
async def get_badges(self) -> dict[str, 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:
@@ -325,7 +321,7 @@ class AsyncClient(AbstractClient):
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()}
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.
@@ -359,7 +355,7 @@ class AsyncClient(AbstractClient):
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:
@@ -395,7 +391,7 @@ class AsyncClient(AbstractClient):
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.
@@ -429,7 +425,7 @@ class AsyncClient(AbstractClient):
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.