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

get flake8 passing

Browse Source
This commit is contained in:
Andrew Herrington
2021-12-06 18:59:50 -06:00
parent 4d1681ea8e
commit 90233c568e
8 changed files with 85 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.flake8
View File

@@ -1,7 +1,7 @@
[flake8]
select = B,B9,C,D,DAR,E,F,N,RST,S,W
ignore = E203,E501,RST201,RST203,RST301,W503
max-line-length = 80
ignore = E203,E501,RST201,RST203,RST301,W503,B902,N805
max-line-length = 119
max-complexity = 10
docstring-convention = google
per-file-ignores = tests/*:S101

1
.pre-commit-config.yaml
View File

@@ -32,6 +32,7 @@ repos:
entry: flake8
language: system
types: [python]
exclude: "tests/*"
require_serial: true
- id: pyupgrade
name: pyupgrade

0
src/healthchecks_io/schemas/__init__.py Normal file
View File

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

@@ -1,14 +1,15 @@
"""
Schemas for badges
"""Schemas for badges.
https://healthchecks.io/docs/api/
"""
from typing import Dict
from pydantic import AnyUrl
from pydantic import BaseModel
class Badges(BaseModel):
"""Object with the Badges urls."""
svg: str
svg3: str
json_url: str
@@ -18,9 +19,7 @@ class Badges(BaseModel):
@classmethod
def from_api_result(cls, badges_dict: Dict[str, str]) -> "Badges":
"""
Converts an API response into a Badges object
"""
"""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"]
return cls(**badges_dict)

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

@@ -1,5 +1,5 @@
"""
Schemas for checks
"""Schemas for checks.
https://healthchecks.io/docs/api/
"""
from datetime import datetime
@@ -13,13 +13,14 @@ from urllib.parse import urlparse
import pytz
from croniter import croniter
from pydantic import AnyUrl
from pydantic import BaseModel
from pydantic import Field
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: Optional[str]
name: str
slug: str
@@ -42,10 +43,9 @@ class Check(BaseModel):
@validator("uuid", always=True)
def validate_uuid(
cls, value: Optional[str], values: Dict[str, Any]
cls, value: Optional[str], values: Dict[str, Any] # noqa: B902
) -> Optional[str]:
"""
Tries to set the uuid from the ping_url.
"""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
"""
@@ -58,13 +58,13 @@ class Check(BaseModel):
@classmethod
def from_api_result(cls, check_dict: Dict[str, Any]) -> "Check":
"""
Converts a dict result from the healthchecks API into a Check object
"""
"""Converts a dictionary from the healthchecks api into an Check object."""
return cls(**check_dict)
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"
@@ -84,43 +84,68 @@ class CheckCreate(BaseModel):
)
schedule: Optional[str] = Field(
"* * * * *",
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.",
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: 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,
description="Controls whether a paused check automatically resumes when pinged (the default) or not. If set to false, a paused check will leave the paused state when it receives a ping. If set to true, a paused check will ignore pings and stay paused until you manually resume it from the web dashboard.",
description="Controls whether a paused check automatically resumes "
"when pinged (the default) or not. If set to false, a paused "
"check will leave the paused state when it receives a ping. "
"If set to true, a paused check will ignore pings and stay "
"paused until you manually resume it from the web dashboard.",
)
methods: Optional[str] = Field(
"",
description="Specifies the allowed HTTP methods for making ping requests. Must be one of the two values: an empty string or POST. Set this field to an empty string to allow HEAD, GET, and POST requests. Set this field to POST to allow only POST requests.",
description="Specifies the allowed HTTP methods for making "
"ping requests. Must be one of the two values: an empty "
"string or POST. Set this field to an empty string to "
"allow HEAD, GET, and POST requests. Set this field to "
"POST to allow only POST requests.",
)
channels: Optional[str] = Field(
None,
description="By default, this API call assigns no integrations to the newly created check. By default, this API call assigns no integrations to the newly created check. To assign specific integrations, use a comma-separated list of integration UUIDs.",
description="By default, this API call assigns no integrations"
"to the newly created check. By default, this API call "
"assigns no integrations to the newly created check. "
"To assign specific integrations, use a comma-separated list "
"of integration UUIDs.",
)
unique: Optional[List[Optional[str]]] = Field(
[],
description="Enables upsert functionality. Before creating a check, Healthchecks.io looks for existing checks, filtered by fields listed in unique. If Healthchecks.io does not find a matching check, it creates a new check and returns it with the HTTP status code 201 If Healthchecks.io finds a matching check, it updates the existing check and returns it with HTTP status code 200. The accepted values for the unique field are name, tags, timeout, and grace.",
description="Enables upsert functionality. Before creating a check, "
"Healthchecks.io looks for existing checks, filtered by fields listed "
"in unique. If Healthchecks.io does not find a matching check, it "
"creates a new check and returns it with the HTTP status code 201 "
"If Healthchecks.io finds a matching check, it updates the existing "
"check and returns it with HTTP status code 200. The accepted values "
"for the unique field are name, tags, timeout, and grace.",
)
@validator("schedule")
def validate_schedule(cls, value: str) -> str:
"""Validates that the schedule is a valid cron expression."""
if not croniter.is_valid(value):
raise ValueError("Schedule is not a valid cron expression")
return value
@validator("tz")
def validate_tz(cls, value: str) -> str:
if not value in pytz.all_timezones:
"""Validates that the timezone is a valid timezone string."""
if value not in pytz.all_timezones:
raise ValueError("Tz is not a valid timezone")
return value
@validator("methods")
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"
@@ -129,6 +154,7 @@ class CheckCreate(BaseModel):
@validator("unique")
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"):
raise ValueError(
@@ -138,6 +164,8 @@ class CheckCreate(BaseModel):
class CheckPings(BaseModel):
"""A Pydantic schema for a check's Pings."""
type: str
date: datetime
number_of_pings: int
@@ -151,11 +179,14 @@ class CheckPings(BaseModel):
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"]
return cls(**ping_dict)
class CheckStatuses(BaseModel):
"""A Pydantic schema for a check's Statuses."""
timestamp: datetime
up: int

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

@@ -1,5 +1,5 @@
"""
Schemas for integrations
"""Schemas for integrations.
https://healthchecks.io/docs/api/
"""
from typing import Dict
@@ -8,10 +8,13 @@ from pydantic import BaseModel
class Integration(BaseModel):
"""Schema for an integration object."""
id: str
name: str
kind: str
@classmethod
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)

14
tests/schemas/test_badges.py Normal file
View File

@@ -0,0 +1,14 @@
from healthchecks_io.schemas.badges import Badges
def test_badge_from_api_result():
badges_dict = {
"svg": "https://healthchecks.io/badge/67541b37-8b9c-4d17-b952-690eae/LOegDs5M-2/backup.svg",
"svg3": "https://healthchecks.io/badge/67541b37-8b9c-4d17-b952-690eae/LOegDs5M/backup.svg",
"json": "https://healthchecks.io/badge/67541b37-8b9c-4d17-b952-690eae/LOegDs5M-2/backup.json",
"json3": "https://healthchecks.io/badge/67541b37-8b9c-4d17-b952-690eae/LOegDs5M/backup.json",
"shields": "https://healthchecks.io/badge/67541b37-8b9c-4d17-b952-690eae/LOegDs5M-2/backup.shields",
"shields3": "https://healthchecks.io/badge/67541b37-8b9c-4d17-b952-690eae/LOegDs5M/backup.shields"
}
this_badge = Badges.from_api_result(badges_dict)
assert this_badge.svg == badges_dict['svg']
assert this_badge.json_url == badges_dict['json']

11
tests/schemas/test_integrations.py Normal file
View File

@@ -0,0 +1,11 @@
from healthchecks_io.schemas.integrations import Integration
def test_badge_from_api_result():
int_dict = {
"id": "4ec5a071-2d08-4baa-898a-eb4eb3cd6941",
"name": "My Work Email",
"kind": "email"
}
this_integration = Integration.from_api_result(int_dict)
assert this_integration.id == int_dict['id']
assert this_integration.name == int_dict['name']