from dataclasses import dataclass
from typing import Any, Callable
from base64 import b64encode, b64decode
+from binascii import Error as binascii_Exception
from http.server import BaseHTTPRequestHandler
from http.server import HTTPServer
from urllib.parse import urlparse, parse_qs
@staticmethod
def _request_wrapper(http_method: str, not_found_msg: str
) -> Callable[..., Callable[[TaskHandler], None]]:
+ """Wrapper for do_GET… and do_POST… handlers, to init and clean up.
+
+ Among other things, conditionally cleans all caches, but only on POST
+ requests, as only those are expected to change the states of objects
+ that may be cached, and certainly only those are expected to write any
+ changes to the database. We want to call them as early though as
+ possible here, either exactly after the specific request handler
+ returns successfully, or right after any exception is triggered –
+ otherwise, race conditions become plausible.
+
+ Note that otherwise any POST attempt, even a failed one, may end in
+ problematic inconsistencies:
+
+ - if the POST handler experiences an Exception, changes to objects
+ won't get written to the DB, but the changed objects may remain in
+ the cache and affect other objects despite their possibly illegal
+ state
+
+ - even if an object was just saved to the DB, we cannot be sure its
+ current state is completely identical to what we'd get if loading it
+ fresh from the DB (e.g. currently Process.n_owners is only updated
+ when loaded anew via .from_table_row, nor is its state written to
+ the DB by .save; a questionable design choice, but proof that we
+ have no guarantee that objects' .save stores all their states we'd
+ prefer at their most up-to-date.
+ """
+
+ def clear_caches() -> None:
+ for cls in (Day, Todo, Condition, Process, ProcessStep):
+ assert hasattr(cls, 'empty_cache')
+ cls.empty_cache()
+
def decorator(f: Callable[..., str | None]
) -> Callable[[TaskHandler], None]:
def wrapper(self: TaskHandler) -> None:
if hasattr(self, handler_name):
handler = getattr(self, handler_name)
redir_target = f(self, handler)
+ if 'POST' == http_method:
+ clear_caches()
if redir_target:
self.send_response(302)
self.send_header('Location', redir_target)
msg = f'{not_found_msg}: {self._site}'
raise NotFoundException(msg)
except HandledException as error:
- for cls in (Day, Todo, Condition, Process, ProcessStep):
- assert hasattr(cls, 'empty_cache')
- cls.empty_cache()
+ if 'POST' == http_method:
+ clear_caches()
ctx = {'msg': error}
self._send_page(ctx, 'msg', error.http_code)
finally:
owned_ids = self._params.get_all_int('has_step')
title_64 = self._params.get_str('title_b64')
if title_64:
- title = b64decode(title_64.encode()).decode()
+ try:
+ title = b64decode(title_64.encode()).decode()
+ except binascii_Exception as exc:
+ msg = 'invalid base64 for ?title_b64='
+ raise BadFormatException(msg) from exc
process.title.set(title)
+ preset_top_step = None
owners = process.used_as_step_by(self.conn)
for step_id in owner_ids:
owners += [Process.by_id(self.conn, step_id)]
- preset_top_step = None
for process_id in owned_ids:
+ Process.by_id(self.conn, process_id) # to ensure ID exists
preset_top_step = process_id
return {'process': process, 'is_new': process.id_ is None,
'preset_top_step': preset_top_step,