From: mar77i Date: Mon, 10 Jun 2024 20:43:37 +0000 (+0200) Subject: add a bunch of typehints X-Git-Url: https://git.mar77i.info/?a=commitdiff_plain;h=f591748910835b1a11c3765723d9d30193a5bd26;p=hublib add a bunch of typehints --- diff --git a/hub/app.py b/hub/app.py index 5192a91..3d70154 100644 --- a/hub/app.py +++ b/hub/app.py @@ -7,6 +7,7 @@ from pathlib import Path from secrets import token_urlsafe from subprocess import run from traceback import print_exception +from typing import Generator from urllib.parse import urlunsplit from falcon.asgi import App as FalconApp @@ -20,12 +21,12 @@ from .static import HubApp, TemplateTreeFileApp class App(TemplateTreeFileApp, FalconApp): @classmethod - def scan_files(cls, base_dir): + def scan_files(cls, base_dir: Path) -> Generator[Path]: for path in base_dir.iterdir(): if not path.is_dir() and not cls.is_ignored_filename(path): yield path - def scramble(self, value): + def scramble(self, value: str | bytes) -> str: if isinstance(value, str): value = value.encode() secret = self.secret @@ -38,7 +39,7 @@ class App(TemplateTreeFileApp, FalconApp): def uri_tail(self, path: Path) -> str: return self.scramble(super().uri_tail(path)) - def __init__(self, base_dir, secret, **kwargs): + def __init__(self, base_dir: Path, secret: str, **kwargs): from .utils import get_redis_pass self.secret = secret or token_urlsafe(64) @@ -63,7 +64,7 @@ class App(TemplateTreeFileApp, FalconApp): ) self.add_error_handler(Exception, self.print_exception) - async def print_exception(self, req, resp, ex, params): + async def print_exception(self, req, resp, ex, params) -> None: print_exception(*sys.exc_info()) diff --git a/hub/static.py b/hub/static.py index 8c50f77..f3e0d84 100644 --- a/hub/static.py +++ b/hub/static.py @@ -1,6 +1,8 @@ from pathlib import Path from falcon.constants import MEDIA_HTML, MEDIA_JS, MEDIA_TEXT +from falcon.request import Request +from falcon.response import Response from falcon.status_codes import HTTP_OK from jinja2 import Template @@ -13,29 +15,28 @@ class StaticFile: """ Basic static file wrapper. """ - media_types_per_suffix = { + media_types_per_suffix: dict[str, str] = { ".html": MEDIA_HTML, ".js": MEDIA_JS, ".css": MEDIA_CSS, ".txt": MEDIA_TEXT, } - content: str | None - def __init__(self, path): + def __init__(self, path: Path): self.path = path self.name = path.name self.media_type = self.get_media_type(path) - self.mtime = None - self.content = None + self.mtime: float | None = None + self.content: bytes | None = None @classmethod - def get_media_type(cls, path): + def get_media_type(cls, path: Path): return cls.media_types_per_suffix[path.suffix] - def get(self): + def get(self) -> bytes: mtime = self.path.stat().st_mtime if mtime != self.mtime: - with open(self.path) as fh: + with open(self.path, "rb") as fh: self.content = fh.read() self.mtime = mtime return self.content @@ -102,17 +103,17 @@ class TreeFileApp: self.base_uri = base_uri.rstrip("/") self.name = name or self.create_name(self.root.base_uri, self.base_uri) assert self.name, self.name - self.files_per_uris = {} + self.files_per_uris: dict[str, StaticFile] = {} for path in self.scan_files(base_dir): static_file = self.get_file(path) uri = self.uri(static_file.path) self.files_per_uris[uri] = static_file root.add_route(uri, self) - async def on_get(self, req, resp): + async def on_get(self, req: Request, resp: Response): resource = self.files_per_uris[req.path] resp.content_type = resource.media_type - resp.data = resource.get().encode() + resp.data = resource.get() resp.status = HTTP_OK @@ -120,13 +121,13 @@ class StaticTemplateFile(StaticFile): TEMPLATE_SUFFIX = ".j2" content: Template | None - def __init__(self, path, hubapp): + def __init__(self, path: Path, hubapp: "TemplateTreeFileApp"): super().__init__(path) self.name = path.stem self.hubapp = hubapp self.context = {"static_file": self} - def get(self) -> str: + def get(self) -> bytes: mtime = self.path.stat().st_mtime if mtime != self.mtime: env = self.hubapp.root.env @@ -134,10 +135,10 @@ class StaticTemplateFile(StaticFile): str(self.path.relative_to(env.loader.searchpath[0])) ) self.mtime = mtime - return self.content.render(self.context) + return self.content.render(self.context).encode() @classmethod - def get_media_type(cls, path): + def get_media_type(cls, path: Path) -> str: assert path.suffix == cls.TEMPLATE_SUFFIX return cls.media_types_per_suffix[path.suffixes[-2]] @@ -160,9 +161,7 @@ class HubApp(TemplateTreeFileApp): super().__init__(app, base_dir, base_uri) self.ws_app = WebSocketApp(self) for suffix in ("client", "master"): - uri = self.uri(Path(f"ws_{suffix}")) - self.files_per_uris[uri] = self.ws_app - app.add_route(uri, self.ws_app, suffix=suffix) + app.add_route(self.uri(Path(f"ws_{suffix}")), self.ws_app, suffix=suffix) @staticmethod def is_master_uri(path: Path) -> bool: diff --git a/hub/utils.py b/hub/utils.py index 3e91e23..d888f15 100644 --- a/hub/utils.py +++ b/hub/utils.py @@ -1,12 +1,14 @@ from jinja2_simple_tags import StandaloneTag -from .static import TreeFileApp +from .static import StaticTemplateFile, TreeFileApp class StaticTag(StandaloneTag): tags = {"static"} @staticmethod - def get_hubapp(static_file, hubapp): + def get_hubapp( + static_file: StaticTemplateFile, hubapp: str | TreeFileApp | None + ) -> TreeFileApp: h = static_file.hubapp if hubapp == "root": return h.root @@ -42,4 +44,4 @@ def get_redis_pass(redis_conf: str) -> str: for line in fh: if line.startswith(prefix): return line[len(prefix) :].rstrip() - return None + raise ValueError("No redis password found") diff --git a/hub/websocket.py b/hub/websocket.py index 89a7e2f..0ce426a 100644 --- a/hub/websocket.py +++ b/hub/websocket.py @@ -8,9 +8,11 @@ from traceback import print_exception from falcon import WebSocketDisconnected +from .static import TreeFileApp + class WebSocketApp: - def __init__(self, hubapp): + def __init__(self, hubapp: TreeFileApp): self.name = hubapp.name self.conn = hubapp.root.conn