diff --git a/.flake8 b/.flake8 index 0f46431..7bc6eae 100644 --- a/.flake8 +++ b/.flake8 @@ -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 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 110c5c8..e727a9c 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -32,6 +32,7 @@ repos: entry: flake8 language: system types: [python] + exclude: "tests/*" require_serial: true - id: pyupgrade name: pyupgrade diff --git a/src/healthchecks_io/schemas/__init__.py b/src/healthchecks_io/schemas/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/healthchecks_io/schemas/badges.py b/src/healthchecks_io/schemas/badges.py index 21237f0..09bb1a5 100644 --- a/src/healthchecks_io/schemas/badges.py +++ b/src/healthchecks_io/schemas/badges.py @@ -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) diff --git a/src/healthchecks_io/schemas/checks.py b/src/healthchecks_io/schemas/checks.py index 93a2340..760c5cd 100644 --- a/src/healthchecks_io/schemas/checks.py +++ b/src/healthchecks_io/schemas/checks.py @@ -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 diff --git a/src/healthchecks_io/schemas/integrations.py b/src/healthchecks_io/schemas/integrations.py index 2530131..b1594e8 100644 --- a/src/healthchecks_io/schemas/integrations.py +++ b/src/healthchecks_io/schemas/integrations.py @@ -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) diff --git a/tests/schemas/test_badges.py b/tests/schemas/test_badges.py new file mode 100644 index 0000000..f195158 --- /dev/null +++ b/tests/schemas/test_badges.py @@ -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'] diff --git a/tests/schemas/test_integrations.py b/tests/schemas/test_integrations.py new file mode 100644 index 0000000..11a829d --- /dev/null +++ b/tests/schemas/test_integrations.py @@ -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']