home · contact · privacy
Re-organize HTTP module code for better readability.
[plomtask] / plomtask / http.py
index 72ad872befe0f663dc6435a5ee8584e66b31a29b..ba79c5a5a929fb4d02975276a854e5b6de4c999b 100644 (file)
@@ -20,16 +20,6 @@ from plomtask.todos import Todo
 TEMPLATES_DIR = 'templates'
 
 
-@dataclass
-class TodoStepsNode:
-    """Collect what's useful for Todo steps tree display."""
-    id_: int
-    todo: Todo | None
-    process: Process | None
-    children: list[TodoStepsNode]
-    fillable: bool = False
-
-
 class TaskServer(HTTPServer):
     """Variant of HTTPServer that knows .jinja as Jinja Environment."""
 
@@ -123,10 +113,10 @@ class TaskHandler(BaseHTTPRequestHandler):
         self.wfile.write(bytes(html, 'utf-8'))
 
     @staticmethod
-    def _wrap(method: str, not_found_msg: str
-              ) -> Callable[..., Callable[..., None]]:
-        def decorator(http_method_handler: Callable[..., str]
-                      ) -> Callable[..., None]:
+    def _request_wrapper(http_method: str, not_found_msg: str
+                         ) -> Callable[..., Callable[[TaskHandler], None]]:
+        def decorator(f: Callable[..., str | None]
+                      ) -> Callable[[TaskHandler], None]:
             def wrapper(self: TaskHandler) -> None:
                 try:
                     self.conn = DatabaseConnection(self.server.db)
@@ -134,10 +124,10 @@ class TaskHandler(BaseHTTPRequestHandler):
                     self.site = path_split(parsed_url.path)[1]
                     params = parse_qs(parsed_url.query, strict_parsing=True)
                     self.params = InputsParser(params, False)
-                    handler_name = f'do_{method}_{self.site}'
+                    handler_name = f'do_{http_method}_{self.site}'
                     if hasattr(self, handler_name):
                         handler = getattr(self, handler_name)
-                        redir_target = http_method_handler(self, handler)
+                        redir_target = f(self, handler)
                         if redir_target:
                             self.send_response(302)
                             self.send_header('Location', redir_target)
@@ -154,8 +144,8 @@ class TaskHandler(BaseHTTPRequestHandler):
             return wrapper
         return decorator
 
-    @_wrap('GET', 'Unknown page')
-    def do_GET(self, handler: Callable[..., str | dict[str, object]]
+    @_request_wrapper('GET', 'Unknown page')
+    def do_GET(self, handler: Callable[[], str | dict[str, object]]
                ) -> str | None:
         """Render page with result of handler, or redirect if result is str."""
         template = f'{self.site}.html'
@@ -167,8 +157,8 @@ class TaskHandler(BaseHTTPRequestHandler):
         self.send_html(html)
         return None
 
-    @_wrap('POST', 'Unknown POST target')
-    def do_POST(self, handler: Callable[..., str]) -> str:
+    @_request_wrapper('POST', 'Unknown POST target')
+    def do_POST(self, handler: Callable[[], str]) -> str:
         """Handle POST with handler, prepare redirection to result."""
         length = int(self.headers['content-length'])
         postvars = parse_qs(self.rfile.read(length).decode(),
@@ -178,8 +168,19 @@ class TaskHandler(BaseHTTPRequestHandler):
         self.conn.commit()
         return redir_target
 
+    # GET handlers
+
+    def do_GET_(self) -> str:
+        """Return redirect target on GET /."""
+        return '/day'
+
     def _do_GET_calendar(self) -> dict[str, object]:
-        """Show Days from ?start= to ?end=."""
+        """Show Days from ?start= to ?end=.
+
+        Both .do_GET_calendar and .do_GET_calendar_txt refer to this to do the
+        same, the only difference being the HTML template they are rendered to,
+        which .do_GET selects from their method name.
+        """
         start = self.params.get_str('start')
         end = self.params.get_str('end')
         if not end:
@@ -192,10 +193,6 @@ class TaskHandler(BaseHTTPRequestHandler):
         today = date_in_n_days(0)
         return {'start': start, 'end': end, 'days': days, 'today': today}
 
-    def do_GET_(self) -> str:
-        """Return redirect target on GET /."""
-        return '/day'
-
     def do_GET_calendar(self) -> dict[str, object]:
         """Show Days from ?start= to ?end= – normal view."""
         return self._do_GET_calendar()
@@ -240,6 +237,15 @@ class TaskHandler(BaseHTTPRequestHandler):
     def do_GET_todo(self) -> dict[str, object]:
         """Show single Todo of ?id=."""
 
+        @dataclass
+        class TodoStepsNode:
+            """Collect what's useful for Todo steps tree display."""
+            id_: int
+            todo: Todo | None
+            process: Process | None
+            children: list[TodoStepsNode]  # pylint: disable=undefined-variable
+            fillable: bool = False
+
         def walk_process_steps(id_: int,
                                process_step_nodes: list[ProcessStepsNode],
                                steps_nodes: list[TodoStepsNode]) -> None:
@@ -438,6 +444,21 @@ class TaskHandler(BaseHTTPRequestHandler):
             processes.sort(key=lambda p: p.title.newest)
         return {'processes': processes, 'sort_by': sort_by, 'pattern': pattern}
 
+    # POST handlers
+
+    def _change_versioned_timestamps(self, cls: Any, attr_name: str) -> str:
+        """Update history timestamps for VersionedAttribute."""
+        id_ = self.params.get_int_or_none('id')
+        item = cls.by_id(self.conn, id_)
+        attr = getattr(item, attr_name)
+        for k, v in self.form_data.get_first_strings_starting('at:').items():
+            old = k[3:]
+            if old[19:] != v:
+                attr.reset_timestamp(old, f'{v}.0')
+        attr.save(self.conn)
+        cls_name = cls.__name__.lower()
+        return f'/{cls_name}_{attr_name}s?id={item.id_}'
+
     def do_POST_day(self) -> str:
         """Update or insert Day of date and Todos mapped to it."""
         date = self.params.get_str('date')
@@ -528,30 +549,17 @@ class TaskHandler(BaseHTTPRequestHandler):
             condition.save(self.conn)
         return f'/todo?id={todo.id_}'
 
-    def _do_POST_versioned_timestamps(self, cls: Any, attr_name: str) -> str:
-        """Update history timestamps for VersionedAttribute."""
-        id_ = self.params.get_int_or_none('id')
-        item = cls.by_id(self.conn, id_)
-        attr = getattr(item, attr_name)
-        for k, v in self.form_data.get_first_strings_starting('at:').items():
-            old = k[3:]
-            if old[19:] != v:
-                attr.reset_timestamp(old, f'{v}.0')
-        attr.save(self.conn)
-        cls_name = cls.__name__.lower()
-        return f'/{cls_name}_{attr_name}s?id={item.id_}'
-
     def do_POST_process_descriptions(self) -> str:
         """Update history timestamps for Process.description."""
-        return self._do_POST_versioned_timestamps(Process, 'description')
+        return self._change_versioned_timestamps(Process, 'description')
 
     def do_POST_process_efforts(self) -> str:
         """Update history timestamps for Process.effort."""
-        return self._do_POST_versioned_timestamps(Process, 'effort')
+        return self._change_versioned_timestamps(Process, 'effort')
 
     def do_POST_process_titles(self) -> str:
         """Update history timestamps for Process.title."""
-        return self._do_POST_versioned_timestamps(Process, 'title')
+        return self._change_versioned_timestamps(Process, 'title')
 
     def do_POST_process(self) -> str:
         """Update or insert Process of ?id= and fields defined in postvars."""
@@ -625,11 +633,11 @@ class TaskHandler(BaseHTTPRequestHandler):
 
     def do_POST_condition_descriptions(self) -> str:
         """Update history timestamps for Condition.description."""
-        return self._do_POST_versioned_timestamps(Condition, 'description')
+        return self._change_versioned_timestamps(Condition, 'description')
 
     def do_POST_condition_titles(self) -> str:
         """Update history timestamps for Condition.title."""
-        return self._do_POST_versioned_timestamps(Condition, 'title')
+        return self._change_versioned_timestamps(Condition, 'title')
 
     def do_POST_condition(self) -> str:
         """Update/insert Condition of ?id= and fields defined in postvars."""