From e32eec3a22d7e823c5763ba718c820adc9418633 Mon Sep 17 00:00:00 2001 From: Christian Heller Date: Wed, 17 Apr 2024 17:35:45 +0200 Subject: [PATCH] Refactor HTTP module. --- plomtask/http.py | 240 ++++++++++++++++++++++------------------------- 1 file changed, 113 insertions(+), 127 deletions(-) diff --git a/plomtask/http.py b/plomtask/http.py index 912b635..cb00825 100644 --- a/plomtask/http.py +++ b/plomtask/http.py @@ -93,185 +93,171 @@ class TaskHandler(BaseHTTPRequestHandler): def do_GET(self) -> None: """Handle any GET request.""" try: - conn, site, params = self._init_handling() - if site in {'calendar', 'day', 'process', 'processes', 'todo', - 'condition', 'conditions'}: - html = getattr(self, f'do_GET_{site}')(conn, params) - elif '' == site: + self._init_handling() + if self.site in {'calendar', 'day', 'process', 'processes', 'todo', + 'condition', 'conditions'}: + template = f'{self.site}.html' + ctx = getattr(self, f'do_GET_{self.site}')() + html = self.server.jinja.get_template(template).render(**ctx) + self._send_html(html) + elif '' == self.site: self._redirect('/day') - return else: - raise NotFoundException(f'Unknown page: /{site}') - self._send_html(html) + raise NotFoundException(f'Unknown page: /{self.site}') except HandledException as error: self._send_msg(error, code=error.http_code) finally: - conn.close() + self.conn.close() - def do_GET_calendar(self, conn: DatabaseConnection, - params: InputsParser) -> str: + def do_GET_calendar(self) -> dict[str, object]: """Show Days from ?start= to ?end=.""" - start = params.get_str('start') - end = params.get_str('end') - days = Day.all(conn, date_range=(start, end), fill_gaps=True) - return self.server.jinja.get_template('calendar.html').render( - days=days, start=start, end=end) - - def do_GET_day(self, conn: DatabaseConnection, - params: InputsParser) -> str: + start = self.params.get_str('start') + end = self.params.get_str('end') + days = Day.all(self.conn, date_range=(start, end), fill_gaps=True) + return {'start': start, 'end': end, 'days': days} + + def do_GET_day(self) -> dict[str, object]: """Show single Day of ?date=.""" - date = params.get_str('date', todays_date()) - day = Day.by_date(conn, date, create=True) - todos = Todo.by_date(conn, date) + date = self.params.get_str('date', todays_date()) conditions_listing = [] - for condition in Condition.all(conn): - enablers = Todo.enablers_for_at(conn, condition, date) - disablers = Todo.disablers_for_at(conn, condition, date) + for condition in Condition.all(self.conn): + enablers = Todo.enablers_for_at(self.conn, condition, date) + disablers = Todo.disablers_for_at(self.conn, condition, date) conditions_listing += [{ 'condition': condition, 'enablers': enablers, 'disablers': disablers}] - return self.server.jinja.get_template('day.html').render( - day=day, processes=Process.all(conn), todos=todos, - conditions_listing=conditions_listing) + return {'day': Day.by_date(self.conn, date, create=True), + 'todos': Todo.by_date(self.conn, date), + 'processes': Process.all(self.conn), + 'conditions_listing': conditions_listing} - def do_GET_todo(self, conn: DatabaseConnection, params: - InputsParser) -> str: + def do_GET_todo(self) -> dict[str, object]: """Show single Todo of ?id=.""" - id_ = params.get_int_or_none('id') - todo = Todo.by_id(conn, id_) - todo_candidates = Todo.by_date(conn, todo.day.date) - return self.server.jinja.get_template('todo.html').render( - todo=todo, todo_candidates=todo_candidates, - condition_candidates=Condition.all(conn)) - - def do_GET_conditions(self, conn: DatabaseConnection, - _: InputsParser) -> str: + id_ = self.params.get_int_or_none('id') + todo = Todo.by_id(self.conn, id_) + return {'todo': todo, + 'todo_candidates': Todo.by_date(self.conn, todo.day.date), + 'condition_candidates': Condition.all(self.conn)} + + def do_GET_conditions(self) -> dict[str, object]: """Show all Conditions.""" - return self.server.jinja.get_template('conditions.html').render( - conditions=Condition.all(conn)) + return {'conditions': Condition.all(self.conn)} - def do_GET_condition(self, conn: DatabaseConnection, - params: InputsParser) -> str: + def do_GET_condition(self) -> dict[str, object]: """Show Condition of ?id=.""" - id_ = params.get_int_or_none('id') - condition = Condition.by_id(conn, id_, create=True) - return self.server.jinja.get_template('condition.html').render( - condition=condition) + id_ = self.params.get_int_or_none('id') + return {'condition': Condition.by_id(self.conn, id_, create=True)} - def do_GET_process(self, conn: DatabaseConnection, - params: InputsParser) -> str: + def do_GET_process(self) -> dict[str, object]: """Show process of ?id=.""" - id_ = params.get_int_or_none('id') - process = Process.by_id(conn, id_, create=True) - owners = process.used_as_step_by(conn) - return self.server.jinja.get_template('process.html').render( - process=process, steps=process.get_steps(conn), - owners=owners, process_candidates=Process.all(conn), - condition_candidates=Condition.all(conn)) - - def do_GET_processes(self, conn: DatabaseConnection, - _: InputsParser) -> str: + id_ = self.params.get_int_or_none('id') + process = Process.by_id(self.conn, id_, create=True) + return {'process': process, + 'steps': process.get_steps(self.conn), + 'owners': process.used_as_step_by(self.conn), + 'process_candidates': Process.all(self.conn), + 'condition_candidates': Condition.all(self.conn)} + + def do_GET_processes(self) -> dict[str, object]: """Show all Processes.""" - return self.server.jinja.get_template('processes.html').render( - processes=Process.all(conn)) + return {'processes': Process.all(self.conn)} def do_POST(self) -> None: """Handle any POST request.""" + # pylint: disable=attribute-defined-outside-init try: - conn, site, params = self._init_handling() + self._init_handling() length = int(self.headers['content-length']) postvars = parse_qs(self.rfile.read(length).decode(), keep_blank_values=True, strict_parsing=True) - # form_data = PostvarsParser(postvars) - form_data = InputsParser(postvars) - if site in ('day', 'process', 'todo', 'condition'): - getattr(self, f'do_POST_{site}')(conn, params, form_data) - conn.commit() + self.form_data = InputsParser(postvars) + if self.site in ('day', 'process', 'todo', 'condition'): + getattr(self, f'do_POST_{self.site}')() + self.conn.commit() else: - msg = f'Page not known as POST target: /{site}' + msg = f'Page not known as POST target: /{self.site}' raise NotFoundException(msg) self._redirect('/') except HandledException as error: self._send_msg(error, code=error.http_code) finally: - conn.close() + self.conn.close() - def do_POST_day(self, conn: DatabaseConnection, params: InputsParser, - form_data: InputsParser) -> None: + def do_POST_day(self) -> None: """Update or insert Day of date and Todos mapped to it.""" - date = params.get_str('date') - day = Day.by_date(conn, date, create=True) - day.comment = form_data.get_str('comment') - day.save(conn) - process_id = form_data.get_int_or_none('new_todo') + date = self.params.get_str('date') + day = Day.by_date(self.conn, date, create=True) + day.comment = self.form_data.get_str('comment') + day.save(self.conn) + process_id = self.form_data.get_int_or_none('new_todo') if process_id is not None: - process = Process.by_id(conn, process_id) + process = Process.by_id(self.conn, process_id) todo = Todo(None, process, False, day) - todo.save(conn) + todo.save(self.conn) - def do_POST_todo(self, conn: DatabaseConnection, params: InputsParser, - form_data: InputsParser) -> None: + def do_POST_todo(self) -> None: """Update Todo and its children.""" - id_ = params.get_int_or_none('id') - todo = Todo.by_id(conn, id_) - child_id = form_data.get_int_or_none('adopt') + id_ = self.params.get_int_or_none('id') + todo = Todo.by_id(self.conn, id_) + child_id = self.form_data.get_int_or_none('adopt') if child_id is not None: - child = Todo.by_id(conn, child_id) + child = Todo.by_id(self.conn, child_id) todo.add_child(child) - todo.set_conditions(conn, form_data.get_all_int('condition')) - todo.set_fulfills(conn, form_data.get_all_int('fulfills')) - todo.set_undoes(conn, form_data.get_all_int('undoes')) - todo.is_done = len(form_data.get_all_str('done')) > 0 - todo.save(conn) + todo.set_conditions(self.conn, self.form_data.get_all_int('condition')) + todo.set_fulfills(self.conn, self.form_data.get_all_int('fulfills')) + todo.set_undoes(self.conn, self.form_data.get_all_int('undoes')) + todo.is_done = len(self.form_data.get_all_str('done')) > 0 + todo.save(self.conn) for condition in todo.fulfills: - condition.save(conn) + condition.save(self.conn) for condition in todo.undoes: - condition.save(conn) + condition.save(self.conn) - def do_POST_process(self, conn: DatabaseConnection, params: InputsParser, - form_data: InputsParser) -> None: + def do_POST_process(self) -> None: """Update or insert Process of ?id= and fields defined in postvars.""" - id_ = params.get_int_or_none('id') - process = Process.by_id(conn, id_, create=True) - process.title.set(form_data.get_str('title')) - process.description.set(form_data.get_str('description')) - process.effort.set(form_data.get_float('effort')) - process.set_conditions(conn, form_data.get_all_int('condition')) - process.set_fulfills(conn, form_data.get_all_int('fulfills')) - process.set_undoes(conn, form_data.get_all_int('undoes')) - process.save_without_steps(conn) + id_ = self.params.get_int_or_none('id') + process = Process.by_id(self.conn, id_, create=True) + process.title.set(self.form_data.get_str('title')) + process.description.set(self.form_data.get_str('description')) + process.effort.set(self.form_data.get_float('effort')) + process.set_conditions(self.conn, + self.form_data.get_all_int('condition')) + process.set_fulfills(self.conn, self.form_data.get_all_int('fulfills')) + process.set_undoes(self.conn, self.form_data.get_all_int('undoes')) + process.save_without_steps(self.conn) assert process.id_ is not None # for mypy process.explicit_steps = [] - for step_id in form_data.get_all_int('steps'): + for step_id in self.form_data.get_all_int('steps'): for step_process_id in\ - form_data.get_all_int(f'new_step_to_{step_id}'): - process.add_step(conn, None, step_process_id, step_id) - if step_id not in form_data.get_all_int('keep_step'): + self.form_data.get_all_int(f'new_step_to_{step_id}'): + process.add_step(self.conn, None, step_process_id, step_id) + if step_id not in self.form_data.get_all_int('keep_step'): continue - step_process_id = form_data.get_int(f'step_{step_id}_process_id') - parent_id = form_data.get_int_or_none(f'step_{step_id}_parent_id') - process.add_step(conn, step_id, step_process_id, parent_id) - for step_process_id in form_data.get_all_int('new_top_step'): - process.add_step(conn, None, step_process_id, None) - process.fix_steps(conn) - - def do_POST_condition(self, conn: DatabaseConnection, params: InputsParser, - form_data: InputsParser) -> None: + step_process_id = self.form_data.get_int( + f'step_{step_id}_process_id') + parent_id = self.form_data.get_int_or_none( + f'step_{step_id}_parent_id') + process.add_step(self.conn, step_id, step_process_id, parent_id) + for step_process_id in self.form_data.get_all_int('new_top_step'): + process.add_step(self.conn, None, step_process_id, None) + process.fix_steps(self.conn) + + def do_POST_condition(self) -> None: """Update/insert Condition of ?id= and fields defined in postvars.""" - id_ = params.get_int_or_none('id') - condition = Condition.by_id(conn, id_, create=True) - condition.title.set(form_data.get_str('title')) - condition.description.set(form_data.get_str('description')) - condition.save(conn) - - def _init_handling(self) -> tuple[DatabaseConnection, str, InputsParser]: - conn = DatabaseConnection(self.server.db) + id_ = self.params.get_int_or_none('id') + condition = Condition.by_id(self.conn, id_, create=True) + condition.title.set(self.form_data.get_str('title')) + condition.description.set(self.form_data.get_str('description')) + condition.save(self.conn) + + def _init_handling(self) -> None: + # pylint: disable=attribute-defined-outside-init + self.conn = DatabaseConnection(self.server.db) parsed_url = urlparse(self.path) - site = path_split(parsed_url.path)[1] - params = InputsParser(parse_qs(parsed_url.query, strict_parsing=True), - False) - return conn, site, params + self.site = path_split(parsed_url.path)[1] + params = parse_qs(parsed_url.query, strict_parsing=True) + self.params = InputsParser(params, False) def _redirect(self, target: str) -> None: self.send_response(302) -- 2.30.2