Added cookies handling and example for it

Author: michalpokusa

adafruit_httpserver/request.py

@@ -221,7 +221,7 @@ def __repr__(self) -> str:
         return f"{class_name}({repr(self._storage)}, files={repr(self.files._storage)})"
 
 
-class Request:
+class Request:  # pylint: disable=too-many-instance-attributes
     """
     Incoming request, constructed from raw incoming bytes.
     It is passed as first argument to all route handlers.
@@ -292,6 +292,7 @@ def __init__(
         self.client_address = client_address
         self.raw_request = raw_request
         self._form_data = None
+        self._cookies = None
 
         if raw_request is None:
             raise ValueError("raw_request cannot be None")
@@ -316,6 +317,36 @@ def body(self) -> bytes:
     def body(self, body: bytes) -> None:
         self.raw_request = self._raw_header_bytes + b"\r\n\r\n" + body
 
+    @staticmethod
+    def _parse_cookies(cookie_header: str) -> None:
+        """Parse cookies from headers."""
+        if cookie_header is None:
+            return {}
+
+        return {
+            name: value.strip('"')
+            for name, value in [
+                cookie.strip().split("=", 1) for cookie in cookie_header.split(";")
+            ]
+        }
+
+    @property
+    def cookies(self) -> Dict[str, str]:
+        """
+        Cookies sent with the request.
+
+        Example::
+
+            request.headers["Cookie"]
+            # "foo=bar; baz=qux; foo=quux"
+
+            request.cookies
+            # {"foo": "quux", "baz": "qux"}
+        """
+        if self._cookies is None:
+            self._cookies = self._parse_cookies(self.headers.get("Cookie"))
+        return self._cookies
+
     @property
     def form_data(self) -> Union[FormData, None]:
         """

adafruit_httpserver/response.py

@@ -60,13 +60,15 @@ def __init__(  # pylint: disable=too-many-arguments
         *,
         status: Union[Status, Tuple[int, str]] = OK_200,
         headers: Union[Headers, Dict[str, str]] = None,
+        cookies: Dict[str, str] = None,
         content_type: str = None,
     ) -> None:
         """
         :param Request request: Request that this is a response to.
         :param str body: Body of response. Defaults to empty string.
         :param Status status: Status code and text. Defaults to 200 OK.
         :param Headers headers: Headers to include in response. Defaults to empty dict.
+        :param Dict[str, str] cookies: Cookies to be sent with the response.
         :param str content_type: Content type of response. Defaults to None.
         """
 
@@ -76,6 +78,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self._headers = (
             headers.copy() if isinstance(headers, Headers) else Headers(headers)
         )
+        self._cookies = cookies.copy() if cookies else {}
         self._content_type = content_type
         self._size = 0
 
@@ -96,6 +99,9 @@ def _send_headers(
         headers.setdefault("Content-Length", content_length)
         headers.setdefault("Connection", "close")
 
+        for cookie_name, cookie_value in self._cookies.items():
+            headers.add("Set-Cookie", f"{cookie_name}={cookie_value}")
+
         for header, value in headers.items():
             if value is not None:
                 response_message_header += f"{header}: {value}\r\n"
@@ -166,6 +172,7 @@ def __init__(  # pylint: disable=too-many-arguments
         *,
         status: Union[Status, Tuple[int, str]] = OK_200,
         headers: Union[Headers, Dict[str, str]] = None,
+        cookies: Dict[str, str] = None,
         content_type: str = None,
         as_attachment: bool = False,
         download_filename: str = None,
@@ -180,6 +187,7 @@ def __init__(  # pylint: disable=too-many-arguments
           server's ``root_path``.
         :param Status status: Status code and text. Defaults to ``200 OK``.
         :param Headers headers: Headers to include in response.
+        :param Dict[str, str] cookies: Cookies to be sent with the response.
         :param str content_type: Content type of response.
         :param bool as_attachment: If ``True``, the file will be sent as an attachment.
         :param str download_filename: Name of the file to send as an attachment.
@@ -193,6 +201,7 @@ def __init__(  # pylint: disable=too-many-arguments
         super().__init__(
             request=request,
             headers=headers,
+            cookies=cookies,
             content_type=content_type,
             status=status,
         )
@@ -293,19 +302,22 @@ def __init__(  # pylint: disable=too-many-arguments
         *,
         status: Union[Status, Tuple[int, str]] = OK_200,
         headers: Union[Headers, Dict[str, str]] = None,
+        cookies: Dict[str, str] = None,
         content_type: str = None,
     ) -> None:
         """
         :param Request request: Request object
         :param Generator body: Generator that yields chunks of data.
         :param Status status: Status object or tuple with code and message.
         :param Headers headers: Headers to be sent with the response.
+        :param Dict[str, str] cookies: Cookies to be sent with the response.
         :param str content_type: Content type of the response.
         """
 
         super().__init__(
             request=request,
             headers=headers,
+            cookies=cookies,
             status=status,
             content_type=content_type,
         )
@@ -352,17 +364,20 @@ def __init__(
         data: Dict[Any, Any],
         *,
         headers: Union[Headers, Dict[str, str]] = None,
+        cookies: Dict[str, str] = None,
         status: Union[Status, Tuple[int, str]] = OK_200,
     ) -> None:
         """
         :param Request request: Request that this is a response to.
         :param dict data: Data to be sent as JSON.
         :param Headers headers: Headers to include in response.
+        :param Dict[str, str] cookies: Cookies to be sent with the response.
         :param Status status: Status code and text. Defaults to 200 OK.
         """
         super().__init__(
             request=request,
             headers=headers,
+            cookies=cookies,
             status=status,
         )
         self._data = data
@@ -398,6 +413,7 @@ def __init__(
         preserve_method: bool = False,
         status: Union[Status, Tuple[int, str]] = None,
         headers: Union[Headers, Dict[str, str]] = None,
+        cookies: Dict[str, str] = None,
     ) -> None:
         """
         By default uses ``permament`` and ``preserve_method`` to determine the ``status`` code to
@@ -415,6 +431,7 @@ def __init__(
         :param bool preserve_method: Whether to preserve the method of the request.
         :param Status status: Status object or tuple with code and message.
         :param Headers headers: Headers to include in response.
+        :param Dict[str, str] cookies: Cookies to be sent with the response.
         """
 
         if status is not None and (permanent or preserve_method):
@@ -428,7 +445,7 @@ def __init__(
             else:
                 status = MOVED_PERMANENTLY_301 if permanent else FOUND_302
 
-        super().__init__(request, status=status, headers=headers)
+        super().__init__(request, status=status, headers=headers, cookies=cookies)
         self._headers.update({"Location": url})
 
     def _send(self) -> None:
@@ -474,14 +491,17 @@ def __init__(  # pylint: disable=too-many-arguments
         self,
         request: Request,
         headers: Union[Headers, Dict[str, str]] = None,
+        cookies: Dict[str, str] = None,
     ) -> None:
         """
         :param Request request: Request object
         :param Headers headers: Headers to be sent with the response.
+        :param Dict[str, str] cookies: Cookies to be sent with the response.
         """
         super().__init__(
             request=request,
             headers=headers,
+            cookies=cookies,
             content_type="text/event-stream",
         )
         self._headers.setdefault("Cache-Control", "no-cache")
@@ -606,11 +626,13 @@ def __init__(  # pylint: disable=too-many-arguments
         self,
         request: Request,
         headers: Union[Headers, Dict[str, str]] = None,
+        cookies: Dict[str, str] = None,
         buffer_size: int = 1024,
     ) -> None:
         """
         :param Request request: Request object
         :param Headers headers: Headers to be sent with the response.
+        :param Dict[str, str] cookies: Cookies to be sent with the response.
         :param int buffer_size: Size of the buffer used to send and receive messages.
         """
         self._check_request_initiates_handshake(request)
@@ -621,6 +643,7 @@ def __init__(  # pylint: disable=too-many-arguments
             request=request,
             status=SWITCHING_PROTOCOLS_101,
             headers=headers,
+            cookies=cookies,
         )
         self._headers.setdefault("Upgrade", "websocket")
         self._headers.setdefault("Connection", "Upgrade")

docs/examples.rst

@@ -185,6 +185,20 @@ return only the first one.
     :emphasize-lines: 32,47,50
     :linenos:
 
+Cookies
+---------------------
+
+You can use cookies to store data on the client side, that will be sent back to the server with every request.
+They are often used to store authentication tokens, session IDs, but also to user preferences e.g. theme.
+
+To access cookies, use ``request.cookies`` dictionary.
+In order to set cookies,  pass ``cookies`` dictionary to ``Response`` constructor.
+
+.. literalinclude:: ../examples/httpserver_cookies.py
+    :caption: examples/httpserver_cookies.py
+    :emphasize-lines: 70,77
+    :linenos:
+
 Chunked response
 ----------------
 

examples/httpserver_cookies.py

@@ -0,0 +1,81 @@
+# SPDX-FileCopyrightText: 2023 Michał Pokusa
+#
+# SPDX-License-Identifier: Unlicense
+
+import socketpool
+import wifi
+
+from adafruit_httpserver import Server, Request, Response, GET
+
+
+pool = socketpool.SocketPool(wifi.radio)
+server = Server(pool, debug=True)
+
+
+THEMES = {
+    "dark": {
+        "background-color": "#1c1c1c",
+        "color": "white",
+        "button-color": "#181818",
+    },
+    "light": {
+        "background-color": "white",
+        "color": "#1c1c1c",
+        "button-color": "white",
+    },
+}
+
+
+def themed_template(user_preferred_theme: str):
+    theme = THEMES[user_preferred_theme]
+
+    return f"""
+    <html>
+        <head>
+            <title>Cookie Example</title>
+            <style>
+                body {{
+                    background-color: {theme['background-color']};
+                    color: {theme['color']};
+                }}
+
+                button {{
+                    background-color: {theme['button-color']};
+                    color: {theme['color']};
+                    border: 1px solid {theme['color']};
+                    padding: 10px;
+                    margin: 10px;
+                }}
+            </style>
+        </head>
+        <body>
+            <a href="/?theme=dark"><button>Dark theme</button></a>
+            <a href="/?theme=light"><button>Light theme</button></a>
+            <br />
+            <p>
+                After changing the theme, close the tab and open again.
+                Notice that theme stays the same.
+            </p>
+        </body>
+    </html>
+    """
+
+
+@server.route("/", GET)
+def themed_from_cookie(request: Request):
+    """
+    Serve a simple themed page, based on the user's cookie.
+    """
+
+    user_theme = request.cookies.get("theme", "light")
+    wanted_theme = request.query_params.get("theme", user_theme)
+
+    return Response(
+        request,
+        themed_template(wanted_theme),
+        content_type="text/html",
+        cookies={} if user_theme == wanted_theme else {"theme": wanted_theme},
+    )
+
+
+server.serve_forever(str(wifi.radio.ipv4_address))