fully typed project
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
from ics import Calendar
|
||||
import utils
|
||||
|
||||
def date_calendar(calendar: Calendar):
|
||||
def date_calendar(calendar: Calendar) -> None:
|
||||
time = utils.format_datetime("%d/%m/%Y %H:%M:%S")
|
||||
|
||||
for event in calendar.events:
|
||||
|
||||
@@ -2,7 +2,7 @@ from ics import Calendar
|
||||
import requests
|
||||
import logger
|
||||
|
||||
def grab_calendar(ics_url: str):
|
||||
def grab_calendar(ics_url: str) -> Calendar | None:
|
||||
response = requests.get(ics_url)
|
||||
|
||||
if response.status_code != 200:
|
||||
|
||||
@@ -1,16 +1,17 @@
|
||||
from http.server import BaseHTTPRequestHandler
|
||||
import urllib.parse as urllib
|
||||
from typing import Callable
|
||||
import logger
|
||||
|
||||
def generate_class(response_callback):
|
||||
def generate_class(response_callback: Callable[[str, str], str | None]) -> type[BaseHTTPRequestHandler]:
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def _send_error(self, code: int, message: str):
|
||||
def _send_error(self, code: int, message: str) -> None:
|
||||
self.send_response(code)
|
||||
self.send_header("Content-Type", "text/plain")
|
||||
self.end_headers()
|
||||
self.wfile.write(message.encode("utf-8"))
|
||||
|
||||
def _handle_request(self):
|
||||
def _handle_request(self) -> None:
|
||||
host = self.client_address[0]
|
||||
method = self.command
|
||||
url = urllib.urlparse(self.path)
|
||||
@@ -39,25 +40,25 @@ def generate_class(response_callback):
|
||||
self.end_headers()
|
||||
self.wfile.write(response.encode("utf-8"))
|
||||
|
||||
def do_GET(self):
|
||||
def do_GET(self) -> None:
|
||||
self._handle_request()
|
||||
|
||||
def do_POST(self):
|
||||
def do_POST(self) -> None:
|
||||
self._handle_request()
|
||||
|
||||
def do_PUT(self):
|
||||
def do_PUT(self) -> None:
|
||||
self._handle_request()
|
||||
|
||||
def do_DELETE(self):
|
||||
def do_DELETE(self) -> None:
|
||||
self._handle_request()
|
||||
|
||||
def do_PATCH(self):
|
||||
def do_PATCH(self) -> None:
|
||||
self._handle_request()
|
||||
|
||||
def do_OPTIONS(self):
|
||||
def do_OPTIONS(self) -> None:
|
||||
self._handle_request()
|
||||
|
||||
def log_message(self, format, *args):
|
||||
def log_message(self, format: str, *args) -> None:
|
||||
return
|
||||
|
||||
return Handler
|
||||
@@ -4,17 +4,17 @@ import utils
|
||||
|
||||
_console_lock = threading.Lock()
|
||||
|
||||
def _log(prefix: str, message: str):
|
||||
def _log(prefix: str, message: str) -> None:
|
||||
time = utils.format_datetime("%H:%M:%S")
|
||||
with _console_lock:
|
||||
print(f"[{time}] {prefix} {message}")
|
||||
|
||||
def info(message: str):
|
||||
def info(message: str) -> None:
|
||||
_log("INFO", message)
|
||||
|
||||
def warning(message: str):
|
||||
def warning(message: str) -> None:
|
||||
_log("WARNING", message)
|
||||
|
||||
def error(message: str):
|
||||
def error(message: str) -> None:
|
||||
_log("ERROR", message)
|
||||
sys.exit(1)
|
||||
@@ -5,7 +5,7 @@ import logger
|
||||
import utils
|
||||
from server import Server
|
||||
|
||||
def _handle_exit(sig, frame):
|
||||
def _handle_exit(sig, frame) -> None:
|
||||
print("Exiting...")
|
||||
sys.exit(0)
|
||||
|
||||
@@ -17,13 +17,14 @@ def _get_environment_variable(key: str) -> str:
|
||||
|
||||
return value
|
||||
|
||||
def _parse_environment_variable(key: str, value: str) -> int | None:
|
||||
def _parse_environment_variable(key: str, value: str) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except ValueError:
|
||||
logger.error(f"Environment variable '{key}' must be an integer")
|
||||
return -1
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
signal.signal(signal.SIGTERM, _handle_exit)
|
||||
signal.signal(signal.SIGINT, _handle_exit)
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@ class Server:
|
||||
self._cached_calendar = None
|
||||
self._cached_time = -cache_duration
|
||||
|
||||
def _cache_calendar(self):
|
||||
def _cache_calendar(self) -> bool:
|
||||
if not self._cached_calendar or self._cached_time + self._cache_duration < time.time():
|
||||
calendar = grabber.grab_calendar(self._ics_url)
|
||||
if not calendar:
|
||||
@@ -29,7 +29,7 @@ class Server:
|
||||
|
||||
return True
|
||||
|
||||
def _response_callback(self, td_group: str, sh_group: str):
|
||||
def _response_callback(self, td_group: str, sh_group: str) -> str | None:
|
||||
if not self._cache_calendar():
|
||||
return None
|
||||
|
||||
@@ -37,7 +37,7 @@ class Server:
|
||||
|
||||
return filtered_calendar.serialize()
|
||||
|
||||
def serve(self):
|
||||
def serve(self) -> None:
|
||||
handler_class = handler.generate_class(self._response_callback)
|
||||
server = HTTPServer((self._host, self._port), handler_class)
|
||||
logger.info(f"Listening on {self._host}:{self._port}")
|
||||
|
||||
@@ -3,9 +3,9 @@ from zoneinfo import ZoneInfo
|
||||
|
||||
_time_zone = ""
|
||||
|
||||
def set_time_zone(time_zone: str):
|
||||
def set_time_zone(time_zone: str) -> None:
|
||||
global _time_zone
|
||||
_time_zone = time_zone
|
||||
|
||||
def format_datetime(format: str):
|
||||
def format_datetime(format: str) -> str:
|
||||
return datetime.now(ZoneInfo(_time_zone)).strftime(format)
|
||||
Reference in New Issue
Block a user