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

chore: package updates and fix ci

Browse Source
This commit is contained in:
Andrew Herrington
2024-05-05 13:39:25 -05:00
parent 4ca01eaf0c
commit 778932a2f8
23 changed files with 749 additions and 956 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/healthchecks_io/__init__.py
View File

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

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

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

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

@@ -57,9 +57,7 @@ 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: Optional[Dict[str, Any]] = 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:
@@ -165,9 +163,7 @@ 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
@@ -208,21 +204,15 @@ 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, Union[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
@@ -255,12 +245,7 @@ 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)}"

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

@@ -1,4 +1,5 @@
"""An async healthchecks.io client."""
import asyncio
from types import TracebackType
from typing import Dict

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

@@ -1,4 +1,5 @@
"""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
@@ -68,9 +69,7 @@ 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])
@@ -96,9 +95,7 @@ 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))
@@ -125,9 +122,7 @@ 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])

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

@@ -1,4 +1,5 @@
"""An async healthchecks.io client."""
from types import TracebackType
from typing import Dict
from typing import List

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

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

1
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

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

@@ -2,6 +2,7 @@
https://healthchecks.io/docs/api/
"""
from datetime import datetime
from pathlib import PurePath
from typing import Any
@@ -88,8 +89,7 @@ class CheckCreate(BaseModel):
)
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: Optional[bool] = Field(
False,
@@ -184,8 +184,7 @@ class CheckUpdate(CheckCreate):
)
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: Optional[bool] = Field(
None,

1
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