Browse Source

Add more type hints

pull/711/head
Pēteris Caune 2 years ago
parent
commit
599b4d2e52
No known key found for this signature in database GPG Key ID: E28D7679E9A9EDE2
  1. 58
      hc/api/models.py
  2. 5
      hc/front/views.py

58
hc/api/models.py

@ -6,6 +6,7 @@ import json
import socket
import sys
import time
from typing import Dict, List, Optional, Union
import uuid
from cronsim import CronSim
@ -77,11 +78,9 @@ PO_PRIORITIES = {
}
def isostring(dt):
def isostring(dt) -> Optional[str]:
"""Convert the datetime to ISO 8601 format with no microseconds."""
if dt:
return dt.replace(microsecond=0).isoformat()
return dt.replace(microsecond=0).isoformat() if dt else None
class Check(models.Model):
@ -127,13 +126,13 @@ class Check(models.Model):
def __str__(self):
return "%s (%d)" % (self.name or self.code, self.id)
def name_then_code(self):
def name_then_code(self) -> str:
if self.name:
return self.name
return str(self.code)
def url(self):
def url(self) -> Optional[str]:
"""Return check's ping url in user's preferred style.
Note: this method reads self.project. If project is not loaded already,
@ -151,37 +150,38 @@ class Check(models.Model):
return settings.PING_ENDPOINT + str(self.code)
def details_url(self):
def details_url(self) -> str:
return settings.SITE_ROOT + reverse("hc-details", args=[self.code])
def cloaked_url(self):
def cloaked_url(self) -> str:
return settings.SITE_ROOT + reverse("hc-uncloak", args=[self.unique_key])
def email(self):
def email(self) -> str:
return "%s@%s" % (self.code, settings.PING_EMAIL_DOMAIN)
def clamped_last_duration(self):
def clamped_last_duration(self) -> Optional[td]:
if self.last_duration and self.last_duration < MAX_DELTA:
return self.last_duration
return None
def set_name_slug(self, name):
def set_name_slug(self, name) -> None:
self.name = name
self.slug = slugify(name)
def get_grace_start(self, with_started=True):
def get_grace_start(self, with_started=True) -> Optional[datetime]:
"""Return the datetime when the grace period starts.
If the check is currently new, paused or down, return None.
"""
# NEVER is a constant sentinel value (year 3000).
# Using None instead would make the min() logic clunky.
result = NEVER
if self.kind == "simple" and self.status == "up":
assert self.last_ping is not None
result = self.last_ping + self.timeout
elif self.kind == "cron" and self.status == "up":
assert self.last_ping is not None
# The complex case, next ping is expected based on cron schedule.
# Don't convert to naive datetimes (and so avoid ambiguities around
# DST transitions). cronsim will handle the timezone-aware datetimes.
@ -191,24 +191,22 @@ class Check(models.Model):
if with_started and self.last_start and self.status != "down":
result = min(result, self.last_start)
if result != NEVER:
return result
return result if result != NEVER else None
def going_down_after(self):
def going_down_after(self) -> Optional[datetime]:
"""Return the datetime when the check goes down.
If the check is new or paused, and not currently running, return None.
If the check is already down, also return None.
"""
grace_start = self.get_grace_start()
if grace_start is not None:
return grace_start + self.grace
def get_status(self, with_started=False):
"""Return current status for display."""
return None
def get_status(self, with_started=False) -> str:
"""Return current status for display."""
frozen_now = now()
if self.last_start:
@ -221,6 +219,7 @@ class Check(models.Model):
return self.status
grace_start = self.get_grace_start(with_started=with_started)
assert grace_start is not None
grace_end = grace_start + self.grace
if frozen_now >= grace_end:
return "down"
@ -230,17 +229,17 @@ class Check(models.Model):
return "up"
def assign_all_channels(self):
def assign_all_channels(self) -> None:
channels = Channel.objects.filter(project=self.project)
self.channel_set.set(channels)
def tags_list(self):
def tags_list(self) -> List[str]:
return [t.strip() for t in self.tags.split(" ") if t.strip()]
def matches_tag_set(self, tag_set):
def matches_tag_set(self, tag_set) -> bool:
return tag_set.issubset(self.tags_list())
def channels_str(self):
def channels_str(self) -> str:
"""Return a comma-separated string of assigned channel codes."""
# Is this an unsaved instance?
@ -253,13 +252,12 @@ class Check(models.Model):
return ",".join(sorted(codes))
@property
def unique_key(self):
def unique_key(self) -> str:
code_half = self.code.hex[:16]
return hashlib.sha1(code_half.encode()).hexdigest()
def to_dict(self, readonly=False):
result = {
def to_dict(self, readonly=False) -> Dict[str, Union[str, int, None]]:
result: Dict[str, Union[str, int, None]] = {
"name": self.name,
"slug": self.slug,
"tags": self.tags,
@ -358,7 +356,7 @@ class Check(models.Model):
if self.n_pings % 100 == 0:
self.prune()
def prune(self):
def prune(self) -> None:
"""Remove old pings and notifications."""
threshold = self.n_pings - self.project.owner_profile.ping_log_limit

5
hc/front/views.py

@ -6,6 +6,7 @@ import os
import re
from secrets import token_urlsafe
import sqlite3
import sys
from urllib.parse import urlencode, urlparse
from cron_descriptor import ExpressionDescriptor
@ -53,9 +54,9 @@ from hc.lib import curl, jsonschema
from hc.lib.badges import get_badge_url
from hc.lib.tz import all_timezones
try:
if sys.version_info >= (3, 9):
from zoneinfo import ZoneInfo
except ImportError:
else:
from backports.zoneinfo import ZoneInfo

Loading…
Cancel
Save