web.py 8.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216
  1. from dataclasses import dataclass
  2. import flask
  3. import pystache
  4. from typing import Optional, TypeVar, Type
  5. import lc.config as c
  6. import lc.error as e
  7. import lc.model as m
  8. import lc.request as r
  9. import lc.view as v
  10. T = TypeVar("T", bound=r.Request)
  11. @dataclass
  12. class ApiOK:
  13. response: dict
  14. class Endpoint:
  15. __slots__ = ("user",)
  16. def __init__(self):
  17. self.user = None
  18. # try finding the token
  19. token = None
  20. # first check the HTTP headers
  21. if (auth := flask.request.headers.get("Authorization", None)) :
  22. token = auth.split()[1]
  23. # if that fails, check the session
  24. elif flask.session.get("auth", None):
  25. token = flask.session["auth"]
  26. if token is None:
  27. return
  28. # if that exists and we can deserialize it, then make sure
  29. # it contains a valid user password, too
  30. try:
  31. payload = c.app.load_token(token)
  32. except:
  33. # TODO: be more specific about what errors we're catching
  34. # here!
  35. return
  36. if "name" not in payload:
  37. return
  38. try:
  39. u = m.User.by_slug(payload["name"])
  40. self.user = u
  41. except e.LCException:
  42. return
  43. @staticmethod
  44. def just_get_user() -> Optional[m.User]:
  45. try:
  46. return Endpoint().user
  47. except:
  48. # this is going to catch everything on the off chance that
  49. # there's a bug in the user-validation code: this is used
  50. # in error handlers, so we should be resilient to that!
  51. return None
  52. SHOULD_REDIRECT = set(("application/x-www-form-urlencoded", "multipart/form-data",))
  53. def api_ok(self, redirect: str, data: Optional[dict] = None) -> ApiOK:
  54. if data is None:
  55. data = {"status": "ok"}
  56. data["redirect"] = redirect
  57. content_type = flask.request.content_type or ""
  58. content_type = content_type.split(";")[0]
  59. if content_type in Endpoint.SHOULD_REDIRECT:
  60. raise e.LCRedirect(redirect)
  61. else:
  62. return ApiOK(response=data)
  63. def request_data(self, cls: Type[T]) -> T:
  64. """Construct a Request model from either a JSON payload or a urlencoded payload"""
  65. if flask.request.content_type == "application/json":
  66. try:
  67. return cls.from_json(flask.request.data)
  68. except KeyError as exn:
  69. raise e.BadPayload(key=exn.args[0])
  70. elif flask.request.content_type == "application/x-www-form-urlencoded":
  71. return cls.from_form(flask.request.form)
  72. else:
  73. raise e.BadContentType(flask.request.content_type or "unknown")
  74. def require_authentication(self, name: str) -> m.User:
  75. """
  76. Check that the currently logged-in user exists and is the
  77. same as the user whose username is given. Raises an exception
  78. otherwise.
  79. """
  80. if not self.user or name != self.user.name:
  81. raise e.BadPermissions()
  82. return self.user
  83. def route(self, *args, **kwargs):
  84. """Forward to the appropriate routing method"""
  85. try:
  86. if flask.request.method == "POST":
  87. # all POST methods are "API methods": if we want to
  88. # display information in response to a post, then we
  89. # should redirect to the page where that information
  90. # can be viewed instead of returning that
  91. # information. (I think.)
  92. api_ok = self.api_post(*args, **kwargs) # type: ignore
  93. assert isinstance(api_ok, ApiOK)
  94. return flask.jsonify(api_ok.response)
  95. elif flask.request.method == "DELETE":
  96. return flask.jsonify(self.api_delete(*args, **kwargs).response) # type: ignore
  97. elif (
  98. flask.request.method in ["GET", "HEAD"]
  99. and flask.request.content_type == "application/json"
  100. ):
  101. # Here we're distinguishing between an API GET (i.e. a
  102. # client trying to get JSON data about an endpoint)
  103. # versus a user-level GET (i.e. a user in a browser.)
  104. # I like using the HTTP headers to distinguish these
  105. # cases, while other APIs tend to have a separate /api
  106. # endpoint to do this.
  107. return flask.jsonify(self.api_get(*args, **kwargs).response) # type: ignore
  108. # if an exception arose from an "API method", then we should
  109. # report it as JSON
  110. except e.LCException as exn:
  111. if flask.request.content_type == "application/json":
  112. return ({"status": exn.http_code(), "error": str(exn)}, exn.http_code())
  113. else:
  114. return (self.render_error(exn), exn.http_code())
  115. # also maybe we tried to redirect, so just do that
  116. except e.LCRedirect as exn:
  117. return flask.redirect(exn.to_path())
  118. # if we're here, it means we're just trying to get a typical
  119. # HTML request.
  120. try:
  121. return self.html(*args, **kwargs) # type: ignore
  122. except e.LCException as exn:
  123. return (self.render_error(exn), exn.http_code())
  124. except e.LCRedirect as exn:
  125. return flask.redirect(exn.to_path())
  126. def render_error(self, exn: e.LCException) -> str:
  127. error = v.Error(code=exn.http_code(), message=str(exn))
  128. page = v.Page(title="error", content=render("error", error), user=self.user)
  129. return render("main", page)
  130. # Decorators result in some weird code in Python, especially 'cause it
  131. # doesn't make higher-order functions terse. Let's break this down a
  132. # bit. This out method, `endpoint`, takes the route...
  133. def endpoint(route: str):
  134. """Route an endpoint using our semi-smart routing machinery"""
  135. # but `endpoint` returns another function which is going to be
  136. # called with the result of the definition after it. The argument
  137. # to what we're calling `do_endpoint` here is going to be the
  138. # class object defined afterwards.
  139. def do_endpoint(endpoint_class: Type[Endpoint]):
  140. # we'll just make that explicit here
  141. assert Endpoint in endpoint_class.__bases__
  142. # finally, we need a function that we'll give to Flask in
  143. # order to actually dispatch to. This is the actual routing
  144. # function, which is why it just creates an instance of the
  145. # endpoint provided above and calls the `route` method on it
  146. def func(*args, **kwargs):
  147. return endpoint_class().route(*args, **kwargs)
  148. # use reflection over the methods defined by the endpoint
  149. # class to decide if it needs to accept POST requests or not.
  150. methods = ["GET"]
  151. if "api_post" in dir(endpoint_class):
  152. methods.append("POST")
  153. if "api_delete" in dir(endpoint_class):
  154. methods.append("DELETE")
  155. # this is just for making error messages nicer
  156. func.__name__ = endpoint_class.__name__
  157. # finally, use the Flask routing machinery to register our callback
  158. return c.app.app.route(route, methods=methods)(func)
  159. return do_endpoint
  160. LOADER = pystache.loader.Loader(extension="mustache", search_dirs=["templates"])
  161. def render(name: str, data: Optional[v.View] = None) -> str:
  162. """Load and use a Mustache template from the project root"""
  163. template = LOADER.load_name(name)
  164. renderer = pystache.Renderer(missing_tags="strict", search_dirs=["templates"])
  165. return renderer.render(template, data or {})
  166. @c.app.app.errorhandler(404)
  167. def handle_404(e):
  168. user = Endpoint.just_get_user()
  169. url = flask.request.path
  170. error = v.Error(code=404, message=f"Page {url} not found")
  171. page = v.Page(title="not found", content=render("error", error), user=None)
  172. return render("main", page)
  173. @c.app.app.errorhandler(500)
  174. def handle_500(e):
  175. user = Endpoint.just_get_user()
  176. c.log(f"Internal error: {e}")
  177. error = v.Error(code=500, message=f"An unexpected error occurred")
  178. page = v.Page(title="500", content=render("error", error), user=None)
  179. return render("main", page)