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

Revert "chore: ci fixups and poetry update"

Browse Source 
This reverts commit d0b986025e.
This commit is contained in:
Andrew Herrington
2024-05-05 13:02:14 -05:00
parent d0b986025e
commit 4ca01eaf0c
24 changed files with 1026 additions and 829 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/healthchecks_io/__init__.py
View File

@@ -1,5 +1,4 @@
"""Py Healthchecks.Io."""
# set by poetry-dynamic-versioning
__version__ = "0.4.0" # noqa: E402

1
src/healthchecks_io/client/__init__.py
View File

@@ -1,5 +1,4 @@
"""healthchecks_io clients."""
from .async_client import AsyncClient # noqa: F401
from .check_trap import CheckTrap # noqa: F401
from .sync_client import Client # noqa: F401

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

@@ -1,6 +1,9 @@
from abc import ABC
from abc import abstractmethod
from typing import Any
from typing import Dict
from typing import Optional
from typing import Union
from urllib.parse import parse_qsl
from urllib.parse import ParseResult
from urllib.parse import unquote
@@ -54,7 +57,9 @@ class AbstractClient(ABC):
"""Finalizer method is called by weakref.finalize when the object is dereferenced to do cleanup of clients."""
pass
def _get_api_request_url(self, path: str, params: dict[str, Any] | None = None) -> str:
def _get_api_request_url(
self, path: str, params: Optional[Dict[str, Any]] = None
) -> str:
"""Get a full request url for the healthchecks api.
Args:
@@ -160,7 +165,9 @@ class AbstractClient(ABC):
raise CheckNotFoundError(f"CHeck not found at {response.request.url}")
if response.status_code == 400:
raise BadAPIRequestError(f"Bad request when requesting {response.request.url}. {response.text}")
raise BadAPIRequestError(
f"Bad request when requesting {response.request.url}. {response.text}"
)
return response
@@ -201,15 +208,21 @@ class AbstractClient(ABC):
raise HCAPIRateLimitError(f"Rate limited on {response.request.url}")
if response.status_code == 400:
raise BadAPIRequestError(f"Bad request when requesting {response.request.url}. {response.text}")
raise BadAPIRequestError(
f"Bad request when requesting {response.request.url}. {response.text}"
)
if response.status_code == 409:
raise NonUniqueSlugError(f"Bad request, slug conflict {response.request.url}. {response.text}")
raise NonUniqueSlugError(
f"Bad request, slug conflict {response.request.url}. {response.text}"
)
return response
@staticmethod
def _add_url_params(url: str, params: dict[str, str | int | bool], replace: bool = True) -> str:
def _add_url_params(
url: str, params: Dict[str, Union[str, int, bool]], replace: bool = True
) -> str:
"""Add GET params to provided URL being aware of existing.
:param url: string of target URL
@@ -242,7 +255,12 @@ class AbstractClient(ABC):
# get all the duplicated keys from params and urlencode them, we'll concat this to the params string later
duplicated_params = [x for x in params if x in parsed_get_args]
# get all the args that aren't duplicated and add them to parsed_get_args
parsed_get_args.update({key: parsed_params[key] for key in [x for x in params if x not in parsed_get_args]})
parsed_get_args.update(
{
key: parsed_params[key]
for key in [x for x in params if x not in parsed_get_args]
}
)
# if we have any duplicated parameters, urlencode them, we append them later
extra_parameters = (
f"&{urlencode({key: params[key] for key in duplicated_params}, doseq=True)}"

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

@@ -1,7 +1,11 @@
"""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
@@ -25,7 +29,7 @@ class AsyncClient(AbstractClient):
api_url: str = "https://healthchecks.io/api/",
ping_url: str = "https://hc-ping.com/",
api_version: int = 1,
client: HTTPXAsyncClient | None = None,
client: Optional[HTTPXAsyncClient] = None,
) -> None:
"""An AsyncClient can be used in code using asyncio to work with the Healthchecks.io api.
@@ -60,9 +64,9 @@ class AsyncClient(AbstractClient):
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
traceback: TracebackType | None,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
"""Context manager exit."""
await self._afinalizer_method()
@@ -114,7 +118,7 @@ class AsyncClient(AbstractClient):
)
return Check.from_api_result(response.json())
async def get_checks(self, tags: list[str] | None = None) -> list[Check]:
async def get_checks(self, tags: Optional[List[str]] = None) -> List[Check]:
"""Get a list of checks from the healthchecks api.
Args:
@@ -209,7 +213,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),
@@ -237,10 +241,10 @@ class AsyncClient(AbstractClient):
async def get_check_flips(
self,
check_id: str,
seconds: int | None = None,
start: int | None = None,
end: int | None = None,
) -> list[CheckStatuses]:
seconds: Optional[int] = None,
start: Optional[int] = None,
end: Optional[int] = 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").
@@ -277,7 +281,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[Integration | None]:
async def get_integrations(self) -> List[Optional[Integration]]:
"""Returns a list of integrations belonging to the project.
Raises:
@@ -293,7 +297,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:
@@ -321,7 +325,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.
@@ -355,7 +359,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:
@@ -391,7 +395,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.
@@ -425,7 +429,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.

39
src/healthchecks_io/client/check_trap.py
View File

@@ -1,6 +1,9 @@
"""CheckTrap is a context manager to wrap around python code to communicate results to a Healthchecks check."""
from types import TracebackType
from typing import List
from typing import Optional
from typing import Type
from typing import Union
from .async_client import AsyncClient
from .exceptions import PingFailedError
@@ -13,7 +16,7 @@ class CheckTrap:
def __init__(
self,
client: Client | AsyncClient,
client: Union[Client, AsyncClient],
uuid: str = "",
slug: str = "",
suppress_exceptions: bool = False,
@@ -31,10 +34,10 @@ class CheckTrap:
"""
if uuid == "" and slug == "":
raise Exception("Must pass a slug or an uuid")
self.client: Client | AsyncClient = client
self.client: Union[Client, AsyncClient] = client
self.uuid: str = uuid
self.slug: str = slug
self.log_lines: list[str] = list()
self.log_lines: List[str] = list()
self.suppress_exceptions: bool = suppress_exceptions
def add_log(self, line: str) -> None:
@@ -65,7 +68,9 @@ class CheckTrap:
CheckTrap: self
"""
if isinstance(self.client, AsyncClient):
raise WrongClientError("You passed an AsyncClient, use this as an async context manager")
raise WrongClientError(
"You passed an AsyncClient, use this as an async context manager"
)
result = self.client.start_ping(uuid=self.uuid, slug=self.slug)
if not result[0]:
raise PingFailedError(result[1])
@@ -73,10 +78,10 @@ class CheckTrap:
def __exit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
traceback: TracebackType | None,
) -> bool | None:
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> Optional[bool]:
"""Exit the context manager.
If there is an exception, add it to any log lines and send a fail ping.
@@ -91,7 +96,9 @@ class CheckTrap:
Optional[bool]: self.suppress_exceptions, if true will not raise any exceptions
"""
if exc_type is None:
self.client.success_ping(self.uuid, self.slug, data="\n".join(self.log_lines))
self.client.success_ping(
self.uuid, self.slug, data="\n".join(self.log_lines)
)
else:
self.add_log(str(exc))
self.add_log(str(traceback))
@@ -118,7 +125,9 @@ class CheckTrap:
CheckTrap: self
"""
if isinstance(self.client, Client):
raise WrongClientError("You passed a sync Client, use this as a regular context manager")
raise WrongClientError(
"You passed a sync Client, use this as a regular context manager"
)
result = await self.client.start_ping(self.uuid, self.slug)
if not result[0]:
raise PingFailedError(result[1])
@@ -126,10 +135,10 @@ class CheckTrap:
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
traceback: TracebackType | None,
) -> bool | None:
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> Optional[bool]:
"""Exit the context manager.
If there is an exception, add it to any log lines and send a fail ping.

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

@@ -1,6 +1,10 @@
"""An async healthchecks.io client."""
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 Client as HTTPXClient
@@ -23,7 +27,7 @@ class Client(AbstractClient):
api_url: str = "https://healthchecks.io/api/",
ping_url: str = "https://hc-ping.com/",
api_version: int = 1,
client: HTTPXClient | None = None,
client: Optional[HTTPXClient] = None,
) -> None:
"""An AsyncClient can be used in code using asyncio to work with the Healthchecks.io api.
@@ -58,9 +62,9 @@ class Client(AbstractClient):
def __exit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
traceback: TracebackType | None,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
"""Context manager exit."""
self._finalizer_method()
@@ -69,7 +73,7 @@ class Client(AbstractClient):
"""Closes the httpx client."""
self._client.close()
def get_checks(self, tags: list[str] | None = None) -> list[checks.Check]:
def get_checks(self, tags: Optional[List[str]] = None) -> List[checks.Check]:
"""Get a list of checks from the healthchecks api.
Args:
@@ -204,7 +208,7 @@ class Client(AbstractClient):
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]:
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),
@@ -230,10 +234,10 @@ class Client(AbstractClient):
def get_check_flips(
self,
check_id: str,
seconds: int | None = None,
start: int | None = None,
end: int | None = None,
) -> list[checks.CheckStatuses]:
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").
@@ -268,7 +272,7 @@ class Client(AbstractClient):
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[integrations.Integration | None]:
def get_integrations(self) -> List[Optional[integrations.Integration]]:
"""Returns a list of integrations belonging to the project.
Raises:
@@ -286,7 +290,7 @@ class Client(AbstractClient):
for integration_dict in response.json()["channels"]
]
def get_badges(self) -> dict[str, badges.Badges]:
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:
@@ -313,7 +317,7 @@ class Client(AbstractClient):
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()}
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.
@@ -347,7 +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:
@@ -383,7 +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.
@@ -417,7 +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.

1
src/healthchecks_io/schemas/__init__.py
View File

@@ -1,5 +1,4 @@
"""Schemas for healthchecks_io."""
from .badges import Badges
from .checks import Check
from .checks import CheckCreate

3
src/healthchecks_io/schemas/badges.py
View File

@@ -2,6 +2,7 @@
https://healthchecks.io/docs/api/
"""
from typing import Dict
from pydantic import BaseModel
@@ -17,7 +18,7 @@ class Badges(BaseModel):
shields3: str
@classmethod
def from_api_result(cls, badges_dict: dict[str, str]) -> "Badges":
def from_api_result(cls, badges_dict: Dict[str, str]) -> "Badges":
"""Converts a dictionary from the healthchecks api into a Badges object."""
badges_dict["json_url"] = badges_dict["json"]
badges_dict["json3_url"] = badges_dict["json3"]

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

@@ -2,10 +2,13 @@
https://healthchecks.io/docs/api/
"""
from datetime import datetime
from pathlib import PurePath
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Union
from urllib.parse import urlparse
import pytz
@@ -18,28 +21,28 @@ from pydantic import validator
class Check(BaseModel):
"""Schema for a check object, either from a readonly api request or a rw api request."""
unique_key: str | None
unique_key: Optional[str]
name: str
slug: str
tags: str | None
desc: str | None
tags: Optional[str]
desc: Optional[str]
grace: int
n_pings: int
status: str
last_ping: datetime | None
next_ping: datetime | None
last_ping: Optional[datetime]
next_ping: Optional[datetime]
manual_resume: bool
methods: str | None
methods: Optional[str]
# healthchecks.io's api doesn't return a scheme so we cant use Pydantic AnyUrl here
ping_url: str | None
update_url: str | None
pause_url: str | None
channels: str | None
timeout: int | None
uuid: str | None
ping_url: Optional[str]
update_url: Optional[str]
pause_url: Optional[str]
channels: Optional[str]
timeout: Optional[int]
uuid: Optional[str]
@validator("uuid", always=True)
def validate_uuid(cls, value: str | None, values: dict[str, Any]) -> str | None: # noqa: B902
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
@@ -53,7 +56,7 @@ class Check(BaseModel):
return value
@classmethod
def from_api_result(cls, check_dict: dict[str, Any]) -> "Check":
def from_api_result(cls, check_dict: Dict[str, Any]) -> "Check":
"""Converts a dictionary from the healthchecks api into an Check object."""
return cls(**check_dict)
@@ -61,33 +64,34 @@ class Check(BaseModel):
class CheckCreate(BaseModel):
"""Pydantic object for creating a check."""
name: str | None = Field("", description="Name of the check")
tags: str | None = Field("", description="String separated list of tags to apply")
desc: str | None = Field("", description="Description of the check")
timeout: int | None = Field(
name: Optional[str] = Field("", description="Name of the check")
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,
description="The expected period of this check in seconds.",
gte=60,
lte=31536000,
)
grace: int | None = Field(
grace: Optional[int] = Field(
3600,
description="The grace period for this check in seconds.",
gte=60,
lte=31536000,
)
schedule: str | None = Field(
schedule: Optional[str] = Field(
None,
description="A cron expression defining this check's schedule. "
"If you specify both timeout and schedule parameters, "
"Healthchecks.io will create a Cron check and ignore the "
"timeout value.",
)
tz: str | None = Field(
tz: Optional[str] = Field(
"UTC",
description="Server's timezone. This setting only has an effect " "in combination with the schedule parameter.",
description="Server's timezone. This setting only has an effect "
"in combination with the schedule parameter.",
)
manual_resume: bool | None = Field(
manual_resume: Optional[bool] = Field(
False,
description="Controls whether a paused check automatically resumes "
"when pinged (the default) or not. If set to false, a paused "
@@ -95,7 +99,7 @@ class CheckCreate(BaseModel):
"If set to true, a paused check will ignore pings and stay "
"paused until you manually resume it from the web dashboard.",
)
methods: str | None = Field(
methods: Optional[str] = Field(
"",
description="Specifies the allowed HTTP methods for making "
"ping requests. Must be one of the two values: an empty "
@@ -103,7 +107,7 @@ class CheckCreate(BaseModel):
"allow HEAD, GET, and POST requests. Set this field to "
"POST to allow only POST requests.",
)
channels: str | None = Field(
channels: Optional[str] = Field(
None,
description="By default, this API call assigns no integrations"
"to the newly created check. By default, this API call "
@@ -111,7 +115,7 @@ class CheckCreate(BaseModel):
"To assign specific integrations, use a comma-separated list "
"of integration UUIDs.",
)
unique: list[str | None] | None = Field(
unique: Optional[List[Optional[str]]] = Field(
[],
description="Enables upsert functionality. Before creating a check, "
"Healthchecks.io looks for existing checks, filtered by fields listed "
@@ -144,7 +148,7 @@ class CheckCreate(BaseModel):
return value
@validator("unique")
def validate_unique(cls, value: list[str | None]) -> list[str | None]:
def validate_unique(cls, value: List[Optional[str]]) -> List[Optional[str]]:
"""Validate unique list."""
for unique in value:
if unique not in ("name", "tags", "timeout", "grace"):
@@ -157,32 +161,33 @@ class CheckCreate(BaseModel):
class CheckUpdate(CheckCreate):
"""Pydantic object for updating a check."""
name: str | None = Field(None, description="Name of the check")
tags: str | None = Field(None, description="String separated list of tags to apply")
timeout: int | None = Field(
name: Optional[str] = Field(None, description="Name of the check")
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.",
gte=60,
lte=31536000,
)
grace: int | None = Field(
grace: Optional[int] = Field(
None,
description="The grace period for this check in seconds.",
gte=60,
lte=31536000,
)
schedule: str | None = Field(
schedule: Optional[str] = Field(
None,
description="A cron expression defining this check's schedule. "
"If you specify both timeout and schedule parameters, "
"Healthchecks.io will create a Cron check and ignore the "
"timeout value.",
)
tz: str | None = Field(
tz: Optional[str] = Field(
None,
description="Server's timezone. This setting only has an effect " "in combination with the schedule parameter.",
description="Server's timezone. This setting only has an effect "
"in combination with the schedule parameter.",
)
manual_resume: bool | None = Field(
manual_resume: Optional[bool] = Field(
None,
description="Controls whether a paused check automatically resumes "
"when pinged (the default) or not. If set to false, a paused "
@@ -190,7 +195,7 @@ class CheckUpdate(CheckCreate):
"If set to true, a paused check will ignore pings and stay "
"paused until you manually resume it from the web dashboard.",
)
methods: str | None = Field(
methods: Optional[str] = Field(
None,
description="Specifies the allowed HTTP methods for making "
"ping requests. Must be one of the two values: an empty "
@@ -198,7 +203,7 @@ class CheckUpdate(CheckCreate):
"allow HEAD, GET, and POST requests. Set this field to "
"POST to allow only POST requests.",
)
channels: str | None = Field(
channels: Optional[str] = Field(
None,
description="By default, this API call assigns no integrations"
"to the newly created check. By default, this API call "
@@ -206,7 +211,7 @@ class CheckUpdate(CheckCreate):
"To assign specific integrations, use a comma-separated list "
"of integration UUIDs.",
)
unique: list[str | None] | None = Field(
unique: Optional[List[Optional[str]]] = Field(
None,
description="Enables upsert functionality. Before creating a check, "
"Healthchecks.io looks for existing checks, filtered by fields listed "
@@ -228,10 +233,10 @@ class CheckPings(BaseModel):
remote_addr: str
method: str
user_agent: str
duration: float | None = None
duration: Optional[float] = None
@classmethod
def from_api_result(cls, ping_dict: dict[str, 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"]

3
src/healthchecks_io/schemas/integrations.py
View File

@@ -2,6 +2,7 @@
https://healthchecks.io/docs/api/
"""
from typing import Dict
from pydantic import BaseModel
@@ -14,6 +15,6 @@ class Integration(BaseModel):
kind: str
@classmethod
def from_api_result(cls, integration_dict: dict[str, str]) -> "Integration":
def from_api_result(cls, integration_dict: Dict[str, str]) -> "Integration":
"""Converts a dictionary from the healthchecks api into an Integration object."""
return cls(**integration_dict)