*args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.db = db_file
- self.headers: list[tuple[str, str]] = []
- self._render_mode = 'html'
- self._jinja = JinjaEnv(loader=JinjaFSLoader(TEMPLATES_DIR))
-
- def set_json_mode(self) -> None:
- """Make server send JSON instead of HTML responses."""
- self._render_mode = 'json'
- self.headers += [('Content-Type', 'application/json')]
-
- @staticmethod
- def ctx_to_json(ctx: dict[str, object], conn: DatabaseConnection) -> str:
- """Render ctx into JSON string."""
-
- def walk_ctx(node: object, references: CtxReferences) -> Any:
- if hasattr(node, 'into_reference'):
- if hasattr(node, 'id_') and node.id_ is not None:
- library_growing[0] = True
- return node.into_reference(references)
- if hasattr(node, 'as_dict'):
- d = node.as_dict
- if '_references' in d:
- own_refs = d['_references']
- if own_refs.update(references):
- library_growing[0] = True
- del d['_references']
- return d
- if isinstance(node, (list, tuple)):
- return [walk_ctx(x, references) for x in node]
- if isinstance(node, dict):
- d = {}
- for k, v in node.items():
- d[k] = walk_ctx(v, references)
- return d
- if isinstance(node, HandledException):
- return str(node)
- return node
-
- models = {}
- for cls in [Day, Process, ProcessStep, Condition, Todo]:
- models[cls.__name__] = cls
- library: dict[str, dict[str | int, object]] = {}
- references = CtxReferences({})
- library_growing = [True]
- while library_growing[0]:
- library_growing[0] = False
- for k, v in ctx.items():
- ctx[k] = walk_ctx(v, references)
- for cls_name, ids in references.d.items():
- if cls_name not in library:
- library[cls_name] = {}
- for id_ in ids:
- cls = models[cls_name]
- assert hasattr(cls, 'can_create_by_id')
- if cls.can_create_by_id:
- assert hasattr(cls, 'by_id_or_create')
- d = cls.by_id_or_create(conn, id_).as_dict
- else:
- assert hasattr(cls, 'by_id')
- d = cls.by_id(conn, id_).as_dict
- del d['_references']
- library[cls_name][id_] = d
- references.d[cls_name] = []
- ctx['_library'] = library
- return json_dumps(ctx)
-
- def render(self,
- ctx: dict[str, object],
- tmpl_name: str,
- conn: DatabaseConnection
- ) -> str:
- """Render ctx according to self._render_mode.."""
- tmpl_name = f'{tmpl_name}.{self._render_mode}'
- if 'html' == self._render_mode:
- template = self._jinja.get_template(tmpl_name)
- return template.render(ctx)
- return self.__class__.ctx_to_json(ctx, conn)
+ self.render_mode = 'html'
+ self.jinja = JinjaEnv(loader=JinjaFSLoader(TEMPLATES_DIR))
class InputsParser:
code: int = 200
) -> None:
"""Send ctx as proper HTTP response."""
- body = self.server.render(ctx, tmpl_name, self.conn)
+ body: str
+ headers: list[tuple[str, str]] = []
+ if 'html' == self.server.render_mode:
+ tmpl = self.server.jinja.get_template(f'{tmpl_name}.html')
+ body = tmpl.render(ctx)
+ else:
+ body = self._ctx_to_json(ctx)
+ headers += [('Content-Type', 'application/json')]
self.send_response(code)
- for header_tuple in self.server.headers:
+ for header_tuple in headers:
self.send_header(*header_tuple)
self.end_headers()
self.wfile.write(bytes(body, 'utf-8'))
+ def _ctx_to_json(self, ctx: dict[str, object]) -> str:
+ """Render ctx into JSON string."""
+
+ def walk_ctx(node: object, references: CtxReferences) -> Any:
+ if hasattr(node, 'into_reference'):
+ if hasattr(node, 'id_') and node.id_ is not None:
+ library_growing[0] = True
+ return node.into_reference(references)
+ if hasattr(node, 'as_dict'):
+ d = node.as_dict
+ if '_references' in d:
+ own_refs = d['_references']
+ if own_refs.update(references):
+ library_growing[0] = True
+ del d['_references']
+ return d
+ if isinstance(node, (list, tuple)):
+ return [walk_ctx(x, references) for x in node]
+ if isinstance(node, dict):
+ d = {}
+ for k, v in node.items():
+ d[k] = walk_ctx(v, references)
+ return d
+ if isinstance(node, HandledException):
+ return str(node)
+ return node
+
+ models = {}
+ for cls in [Day, Process, ProcessStep, Condition, Todo]:
+ models[cls.__name__] = cls
+ library: dict[str, dict[str | int, object]] = {}
+ references = CtxReferences({})
+ library_growing = [True]
+ while library_growing[0]:
+ library_growing[0] = False
+ for k, v in ctx.items():
+ ctx[k] = walk_ctx(v, references)
+ for cls_name, ids in references.d.items():
+ if cls_name not in library:
+ library[cls_name] = {}
+ for id_ in ids:
+ cls = models[cls_name]
+ assert hasattr(cls, 'can_create_by_id')
+ if cls.can_create_by_id:
+ assert hasattr(cls, 'by_id_or_create')
+ d = cls.by_id_or_create(self.conn, id_).as_dict
+ else:
+ assert hasattr(cls, 'by_id')
+ d = cls.by_id(self.conn, id_).as_dict
+ del d['_references']
+ library[cls_name][id_] = d
+ references.d[cls_name] = []
+ ctx['_library'] = library
+ return json_dumps(ctx)
+
@staticmethod
def _request_wrapper(http_method: str, not_found_msg: str
) -> Callable[..., Callable[[TaskHandler], None]]: