home · contact · privacy
Add rump Processes, and to those VersionedAttributes.
[plomtask] / plomtask / http.py
1 """Web server stuff."""
2 from http.server import BaseHTTPRequestHandler
3 from http.server import HTTPServer
4 from urllib.parse import urlparse, parse_qs
5 from os.path import split as path_split
6 from jinja2 import Environment as JinjaEnv, FileSystemLoader as JinjaFSLoader
7 from plomtask.days import Day, todays_date
8 from plomtask.misc import HandledException
9 from plomtask.db import DatabaseConnection
10 from plomtask.processes import Process
11
12 TEMPLATES_DIR = 'templates'
13
14
15 class TaskServer(HTTPServer):
16     """Variant of HTTPServer that knows .jinja as Jinja Environment."""
17
18     def __init__(self, db_file, *args, **kwargs):
19         super().__init__(*args, **kwargs)
20         self.db = db_file
21         self.jinja = JinjaEnv(loader=JinjaFSLoader(TEMPLATES_DIR))
22
23
24 class TaskHandler(BaseHTTPRequestHandler):
25     """Handles single HTTP request."""
26     server: TaskServer
27
28     def do_GET(self):
29         """Handle any GET request."""
30         try:
31             conn, site, params = self._init_handling()
32             if 'calendar' == site:
33                 start = params.get('start', [''])[0]
34                 end = params.get('end', [''])[0]
35                 html = self.do_GET_calendar(conn, start, end)
36             elif 'day' == site:
37                 date = params.get('date', [todays_date()])[0]
38                 html = self.do_GET_day(conn, date)
39             elif 'process' == site:
40                 id_ = params.get('id', [None])[0]
41                 try:
42                     id_ = int(id_) if id_ else None
43                 except ValueError as e:
44                     raise HandledException(f'Bad ?id= value: {id_}') from e
45                 html = self.do_GET_process(conn, id_)
46             elif 'processes' == site:
47                 html = self.do_GET_processes(conn)
48             else:
49                 raise HandledException('Test!')
50             conn.commit()
51             conn.close()
52             self._send_html(html)
53         except HandledException as error:
54             self._send_msg(error)
55
56     def do_GET_calendar(self, conn: DatabaseConnection, start: str, end: str):
57         """Show Days."""
58         days = Day.all(conn, date_range=(start, end), fill_gaps=True)
59         return self.server.jinja.get_template('calendar.html').render(
60                 days=days, start=start, end=end)
61
62     def do_GET_day(self, conn: DatabaseConnection, date: str):
63         """Show single Day."""
64         day = Day.by_date(conn, date, create=True)
65         return self.server.jinja.get_template('day.html').render(day=day)
66
67     def do_GET_process(self, conn: DatabaseConnection, id_: int | None):
68         """Show process of id_."""
69         return self.server.jinja.get_template('process.html').render(
70                 process=Process.by_id(conn, id_, create=True))
71
72     def do_GET_processes(self, conn: DatabaseConnection):
73         """Show all Processes."""
74         return self.server.jinja.get_template('processes.html').render(
75                 processes=Process.all(conn))
76
77     def do_POST(self):
78         """Handle any POST request."""
79         try:
80             conn, site, params = self._init_handling()
81             length = int(self.headers['content-length'])
82             postvars = parse_qs(self.rfile.read(length).decode(),
83                                 keep_blank_values=1)
84             if 'day' == site:
85                 date = params.get('date', [None])[0]
86                 self.do_POST_day(conn, date, postvars)
87             elif 'process' == site:
88                 id_ = params.get('id', [None])[0]
89                 try:
90                     id_ = int(id_) if id_ else None
91                 except ValueError as e:
92                     raise HandledException(f'Bad ?id= value: {id_}') from e
93                 self.do_POST_process(conn, id_, postvars)
94             conn.commit()
95             conn.close()
96             self._redirect('/')
97         except HandledException as error:
98             self._send_msg(error)
99
100     def do_POST_day(self, conn: DatabaseConnection, date: str, postvars: dict):
101         """Update or insert Day of date and fields defined in postvars."""
102         day = Day.by_date(conn, date, create=True)
103         day.comment = postvars['comment'][0]
104         day.save(conn)
105
106     def do_POST_process(self, conn: DatabaseConnection, id_: int | None,
107                         postvars: dict):
108         """Update or insert Process of id_ and fields defined in postvars."""
109         process = Process.by_id(conn, id_, create=True)
110         if process:
111             process.title.set(postvars['title'][0])
112             process.description.set(postvars['description'][0])
113             effort = postvars['effort'][0]
114             try:
115                 process.effort.set(float(effort))
116             except ValueError as e:
117                 raise HandledException(f'Bad effort value: {effort}') from e
118             process.save(conn)
119
120     def _init_handling(self):
121         conn = DatabaseConnection(self.server.db)
122         parsed_url = urlparse(self.path)
123         site = path_split(parsed_url.path)[1]
124         params = parse_qs(parsed_url.query)
125         return conn, site, params
126
127     def _redirect(self, target: str):
128         self.send_response(302)
129         self.send_header('Location', target)
130         self.end_headers()
131
132     def _send_html(self, html: str, code: int = 200):
133         """Send HTML as proper HTTP response."""
134         self.send_response(code)
135         self.end_headers()
136         self.wfile.write(bytes(html, 'utf-8'))
137
138     def _send_msg(self, msg: str, code: int = 400):
139         """Send message in HTML formatting as HTTP response."""
140         html = self.server.jinja.get_template('msg.html').render(msg=msg)
141         self._send_html(html, code)