Added FormData class

Author: michalpokusa

adafruit_httpserver/__init__.py

@@ -51,7 +51,7 @@
     CONNECT,
 )
 from .mime_types import MIMETypes
-from .request import Request
+from .request import FormData, Request
 from .response import (
     Response,
     FileResponse,

adafruit_httpserver/request.py

@@ -8,7 +8,7 @@
 """
 
 try:
-    from typing import Dict, Tuple, Union, TYPE_CHECKING
+    from typing import List, Dict, Tuple, Union, Any, TYPE_CHECKING
     from socket import socket
     from socketpool import SocketPool
 
@@ -22,6 +22,108 @@
 from .headers import Headers
 
 
+class FormData:
+    """
+    Class for parsing and storing form data from POST requests.
+
+    Supports ``application/x-www-form-urlencoded``, ``multipart/form-data`` and ``text/plain``
+    content types.
+
+    Examples::
+
+        form_data = FormData(b"foo=bar&baz=qux", "application/x-www-form-urlencoded")
+
+        # or
+
+        form_data = FormData(b"foo=bar\\r\\nbaz=qux", "text/plain")
+
+        # FormData({"foo": "bar", "baz": "qux"})
+
+        form_data.get("foo") # "bar"
+        form_data["foo"] # "bar"
+        form_data.get("non-existent-key") # None
+        form_data.get_list("baz") # ["qux"]
+        "unknown-key" in form_data # False
+        form_data.fields # ["foo", "baz"]
+    """
+
+    _storage: Dict[str, List[Union[str, bytes]]]
+
+    def __init__(self, data: bytes, content_type: str) -> None:
+        self.content_type = content_type
+        self._storage = {}
+
+        if content_type.startswith("application/x-www-form-urlencoded"):
+            self._parse_x_www_form_urlencoded(data)
+
+        elif content_type.startswith("multipart/form-data"):
+            boundary = content_type.split("boundary=")[1]
+            self._parse_multipart_form_data(data, boundary)
+
+        elif content_type.startswith("text/plain"):
+            self._parse_text_plain(data)
+
+    def _add_field_value(self, field_name: str, value: Union[str, bytes]) -> None:
+        if field_name not in self._storage:
+            self._storage[field_name] = [value]
+        else:
+            self._storage[field_name].append(value)
+
+    def _parse_x_www_form_urlencoded(self, data: bytes) -> None:
+        decoded_data = data.decode()
+
+        for field_name, value in [
+            key_value.split("=", 1) for key_value in decoded_data.split("&")
+        ]:
+            self._add_field_value(field_name, value)
+
+    def _parse_multipart_form_data(self, data: bytes, boundary: str) -> None:
+        blocks = data.split(b"--" + boundary.encode())[1:-1]
+
+        for block in blocks:
+            disposition, content = block.split(b"\r\n\r\n", 1)
+            field_name = disposition.split(b'"', 2)[1].decode()
+            value = content[:-2]
+
+            self._add_field_value(field_name, value)
+
+    def _parse_text_plain(self, data: bytes) -> None:
+        lines = data.split(b"\r\n")[:-1]
+
+        for line in lines:
+            field_name, value = line.split(b"=", 1)
+
+            self._add_field_value(field_name.decode(), value.decode())
+
+    def get(self, field_name: str, default: Any = None) -> Union[str, bytes, None]:
+        """Get the value of a field."""
+        return self._storage.get(field_name, [default])[0]
+
+    def get_list(self, field_name: str) -> List[Union[str, bytes]]:
+        """Get the list of values of a field."""
+        return self._storage.get(field_name, [])
+
+    @property
+    def fields(self):
+        """Returns a list of field names."""
+        return list(self._storage.keys())
+
+    def __getitem__(self, field_name: str):
+        return self.get(field_name)
+
+    def __iter__(self):
+        return iter(self._storage)
+
+    def __len__(self):
+        return len(self._storage)
+
+    def __contains__(self, key: str):
+        return key in self._storage
+
+    def __repr__(self) -> str:
+        return f"FormData({repr(self._storage)})"
+
+
 class Request:
     """
     Incoming request, constructed from raw incoming bytes.
@@ -91,6 +193,7 @@ def __init__(
         self.connection = connection
         self.client_address = client_address
         self.raw_request = raw_request
+        self._form_data = None
 
         if raw_request is None:
             raise ValueError("raw_request cannot be None")
@@ -117,6 +220,13 @@ def body(self) -> bytes:
     def body(self, body: bytes) -> None:
         self.raw_request = self._raw_header_bytes + b"\r\n\r\n" + body
 
+    @property
+    def form_data(self) -> Union[FormData, None]:
+        """POST data of the request"""
+        if self._form_data is None and self.method == "POST":
+            self._form_data = FormData(self.body, self.headers["Content-Type"])
+        return self._form_data
+
     def json(self) -> Union[dict, None]:
         """Body of the request, as a JSON-decoded dictionary."""
         return json.loads(self.body) if self.body else None