1 """Web server stuff."""
3 from http.server import BaseHTTPRequestHandler
4 from http.server import HTTPServer
5 from urllib.parse import urlparse, parse_qs
6 from os.path import split as path_split
7 from jinja2 import Environment as JinjaEnv, FileSystemLoader as JinjaFSLoader
8 from plomtask.days import Day, todays_date
9 from plomtask.exceptions import HandledException, BadFormatException, \
11 from plomtask.db import DatabaseConnection, DatabaseFile
12 from plomtask.processes import Process
13 from plomtask.conditions import Condition
14 from plomtask.todos import Todo
16 TEMPLATES_DIR = 'templates'
19 class TaskServer(HTTPServer):
20 """Variant of HTTPServer that knows .jinja as Jinja Environment."""
22 def __init__(self, db_file: DatabaseFile,
23 *args: Any, **kwargs: Any) -> None:
24 super().__init__(*args, **kwargs)
26 self.jinja = JinjaEnv(loader=JinjaFSLoader(TEMPLATES_DIR))
30 """Wrapper for validating and retrieving dict-like HTTP inputs."""
32 def __init__(self, dict_: dict[str, list[str]],
33 strictness: bool = True) -> None:
35 self.strict = strictness
37 def get_str(self, key: str, default: str = '',
38 ignore_strict: bool = False) -> str:
39 """Retrieve single/first string value of key, or default."""
40 if key not in self.inputs.keys() or 0 == len(self.inputs[key]):
41 if self.strict and not ignore_strict:
42 raise BadFormatException(f'no value found for key {key}')
44 return self.inputs[key][0]
46 def get_int(self, key: str) -> int:
47 """Retrieve single/first value of key as int, error if empty."""
48 val = self.get_int_or_none(key)
50 raise BadFormatException(f'unexpected empty value for: {key}')
53 def get_int_or_none(self, key: str) -> int | None:
54 """Retrieve single/first value of key as int, return None if empty."""
55 val = self.get_str(key, ignore_strict=True)
60 except ValueError as e:
61 msg = f'cannot int form field value for key {key}: {val}'
62 raise BadFormatException(msg) from e
64 def get_float(self, key: str) -> float:
65 """Retrieve float value of key from self.postvars."""
66 val = self.get_str(key)
69 except ValueError as e:
70 msg = f'cannot float form field value for key {key}: {val}'
71 raise BadFormatException(msg) from e
73 def get_all_str(self, key: str) -> list[str]:
74 """Retrieve list of string values at key."""
75 if key not in self.inputs.keys():
77 return self.inputs[key]
79 def get_all_int(self, key: str) -> list[int]:
80 """Retrieve list of int values at key."""
81 all_str = self.get_all_str(key)
83 return [int(s) for s in all_str if len(s) > 0]
84 except ValueError as e:
85 msg = f'cannot int a form field value for key {key} in: {all_str}'
86 raise BadFormatException(msg) from e
89 class TaskHandler(BaseHTTPRequestHandler):
90 """Handles single HTTP request."""
93 def do_GET(self) -> None:
94 """Handle any GET request."""
96 conn, site, params = self._init_handling()
97 if site in {'calendar', 'day', 'process', 'processes', 'todo',
98 'condition', 'conditions'}:
99 html = getattr(self, f'do_GET_{site}')(conn, params)
101 self._redirect('/day')
104 raise NotFoundException(f'Unknown page: /{site}')
105 self._send_html(html)
106 except HandledException as error:
107 self._send_msg(error, code=error.http_code)
111 def do_GET_calendar(self, conn: DatabaseConnection,
112 params: InputsParser) -> str:
113 """Show Days from ?start= to ?end=."""
114 start = params.get_str('start')
115 end = params.get_str('end')
116 days = Day.all(conn, date_range=(start, end), fill_gaps=True)
117 return self.server.jinja.get_template('calendar.html').render(
118 days=days, start=start, end=end)
120 def do_GET_day(self, conn: DatabaseConnection,
121 params: InputsParser) -> str:
122 """Show single Day of ?date=."""
123 date = params.get_str('date', todays_date())
124 day = Day.by_date(conn, date, create=True)
125 todos = Todo.by_date(conn, date)
126 conditions_listing = []
127 for condition in Condition.all(conn):
128 enablers = Todo.enablers_for_at(conn, condition, date)
129 disablers = Todo.disablers_for_at(conn, condition, date)
130 conditions_listing += [{
131 'condition': condition,
132 'enablers': enablers,
133 'disablers': disablers}]
134 return self.server.jinja.get_template('day.html').render(
135 day=day, processes=Process.all(conn), todos=todos,
136 conditions_listing=conditions_listing)
138 def do_GET_todo(self, conn: DatabaseConnection, params:
139 InputsParser) -> str:
140 """Show single Todo of ?id=."""
141 id_ = params.get_int_or_none('id')
142 todo = Todo.by_id(conn, id_)
143 todo_candidates = Todo.by_date(conn, todo.day.date)
144 return self.server.jinja.get_template('todo.html').render(
145 todo=todo, todo_candidates=todo_candidates,
146 condition_candidates=Condition.all(conn))
148 def do_GET_conditions(self, conn: DatabaseConnection,
149 _: InputsParser) -> str:
150 """Show all Conditions."""
151 return self.server.jinja.get_template('conditions.html').render(
152 conditions=Condition.all(conn))
154 def do_GET_condition(self, conn: DatabaseConnection,
155 params: InputsParser) -> str:
156 """Show Condition of ?id=."""
157 id_ = params.get_int_or_none('id')
158 condition = Condition.by_id(conn, id_, create=True)
159 return self.server.jinja.get_template('condition.html').render(
162 def do_GET_process(self, conn: DatabaseConnection,
163 params: InputsParser) -> str:
164 """Show process of ?id=."""
165 id_ = params.get_int_or_none('id')
166 process = Process.by_id(conn, id_, create=True)
167 owners = process.used_as_step_by(conn)
168 return self.server.jinja.get_template('process.html').render(
169 process=process, steps=process.get_steps(conn),
170 owners=owners, process_candidates=Process.all(conn),
171 condition_candidates=Condition.all(conn))
173 def do_GET_processes(self, conn: DatabaseConnection,
174 _: InputsParser) -> str:
175 """Show all Processes."""
176 return self.server.jinja.get_template('processes.html').render(
177 processes=Process.all(conn))
179 def do_POST(self) -> None:
180 """Handle any POST request."""
182 conn, site, params = self._init_handling()
183 length = int(self.headers['content-length'])
184 postvars = parse_qs(self.rfile.read(length).decode(),
185 keep_blank_values=True, strict_parsing=True)
186 # form_data = PostvarsParser(postvars)
187 form_data = InputsParser(postvars)
188 if site in ('day', 'process', 'todo', 'condition'):
189 getattr(self, f'do_POST_{site}')(conn, params, form_data)
192 msg = f'Page not known as POST target: /{site}'
193 raise NotFoundException(msg)
195 except HandledException as error:
196 self._send_msg(error, code=error.http_code)
200 def do_POST_day(self, conn: DatabaseConnection, params: InputsParser,
201 form_data: InputsParser) -> None:
202 """Update or insert Day of date and Todos mapped to it."""
203 date = params.get_str('date')
204 day = Day.by_date(conn, date, create=True)
205 day.comment = form_data.get_str('comment')
207 process_id = form_data.get_int_or_none('new_todo')
208 if process_id is not None:
209 process = Process.by_id(conn, process_id)
210 todo = Todo(None, process, False, day)
213 def do_POST_todo(self, conn: DatabaseConnection, params: InputsParser,
214 form_data: InputsParser) -> None:
215 """Update Todo and its children."""
216 id_ = params.get_int_or_none('id')
217 todo = Todo.by_id(conn, id_)
218 child_id = form_data.get_int_or_none('adopt')
219 if child_id is not None:
220 child = Todo.by_id(conn, child_id)
221 todo.add_child(child)
222 todo.set_conditions(conn, form_data.get_all_int('condition'))
223 todo.set_fulfills(conn, form_data.get_all_int('fulfills'))
224 todo.set_undoes(conn, form_data.get_all_int('undoes'))
225 todo.is_done = len(form_data.get_all_str('done')) > 0
227 for condition in todo.fulfills:
229 for condition in todo.undoes:
232 def do_POST_process(self, conn: DatabaseConnection, params: InputsParser,
233 form_data: InputsParser) -> None:
234 """Update or insert Process of ?id= and fields defined in postvars."""
235 id_ = params.get_int_or_none('id')
236 process = Process.by_id(conn, id_, create=True)
237 process.title.set(form_data.get_str('title'))
238 process.description.set(form_data.get_str('description'))
239 process.effort.set(form_data.get_float('effort'))
240 process.set_conditions(conn, form_data.get_all_int('condition'))
241 process.set_fulfills(conn, form_data.get_all_int('fulfills'))
242 process.set_undoes(conn, form_data.get_all_int('undoes'))
243 process.save_without_steps(conn)
244 assert process.id_ is not None # for mypy
245 process.explicit_steps = []
246 for step_id in form_data.get_all_int('steps'):
247 for step_process_id in\
248 form_data.get_all_int(f'new_step_to_{step_id}'):
249 process.add_step(conn, None, step_process_id, step_id)
250 if step_id not in form_data.get_all_int('keep_step'):
252 step_process_id = form_data.get_int(f'step_{step_id}_process_id')
253 parent_id = form_data.get_int_or_none(f'step_{step_id}_parent_id')
254 process.add_step(conn, step_id, step_process_id, parent_id)
255 for step_process_id in form_data.get_all_int('new_top_step'):
256 process.add_step(conn, None, step_process_id, None)
257 process.fix_steps(conn)
259 def do_POST_condition(self, conn: DatabaseConnection, params: InputsParser,
260 form_data: InputsParser) -> None:
261 """Update/insert Condition of ?id= and fields defined in postvars."""
262 id_ = params.get_int_or_none('id')
263 condition = Condition.by_id(conn, id_, create=True)
264 condition.title.set(form_data.get_str('title'))
265 condition.description.set(form_data.get_str('description'))
268 def _init_handling(self) -> tuple[DatabaseConnection, str, InputsParser]:
269 conn = DatabaseConnection(self.server.db)
270 parsed_url = urlparse(self.path)
271 site = path_split(parsed_url.path)[1]
272 params = InputsParser(parse_qs(parsed_url.query, strict_parsing=True),
274 return conn, site, params
276 def _redirect(self, target: str) -> None:
277 self.send_response(302)
278 self.send_header('Location', target)
281 def _send_html(self, html: str, code: int = 200) -> None:
282 """Send HTML as proper HTTP response."""
283 self.send_response(code)
285 self.wfile.write(bytes(html, 'utf-8'))
287 def _send_msg(self, msg: Exception, code: int = 400) -> None:
288 """Send message in HTML formatting as HTTP response."""
289 html = self.server.jinja.get_template('msg.html').render(msg=msg)
290 self._send_html(html, code)