Added Headers.get_directive() and Headers.get_parameter()

@to get_parameter

Author: michalpokusa

adafruit_httpserver/headers.py

@@ -8,7 +8,7 @@
 """
 
 try:
-    from typing import Dict, Tuple
+    from typing import Dict, Tuple, Union
 except ImportError:
     pass
 
@@ -56,6 +56,43 @@ def get(self, name: str, default: str = None) -> Union[str, None]:
         """Returns the value for the given header name, or default if not found."""
         return self._storage.get(name.lower(), [None, default])[1]
 
+    def get_directive(self, name: str, default: str = None) -> Union[str, None]:
+        """
+        Returns the main value (directive) for the given header name, or default if not found.
+
+        Example::
+
+            headers = Headers({"Content-Type": "text/html; charset=utf-8"})
+            headers.get_directive("Content-Type")
+            # 'text/html'
+        """
+
+        header_value = self.get(name)
+        if header_value is None:
+            return default
+        return header_value.split(";")[0].strip('" ')
+
+    def get_parameter(
+        self, name: str, parameter: str, default: str = None
+    ) -> Union[str, None]:
+        """
+        Returns the value of the given parameter for the given header name, or default if not found.
+
+        Example::
+
+            headers = Headers({"Content-Type": "text/html; charset=utf-8"})
+            headers.get_parameter("Content-Type", "charset")
+            # 'utf-8'
+        """
+
+        header_value = self.get(name)
+        if header_value is None:
+            return default
+        for header_parameter in header_value.split(";"):
+            if header_parameter.strip().startswith(parameter):
+                return header_parameter.strip().split("=")[1].strip('" ')
+        return default
+
     def setdefault(self, name: str, default: str = None):
         """Sets the value for the given header name if it does not exist."""
         return self._storage.setdefault(name.lower(), [name, default])[1]

adafruit_httpserver/response.py

@@ -581,18 +581,18 @@ def route_func(request: Request):
 
     @staticmethod
     def _check_request_initiates_handshake(request: Request):
-        if any(
+        if not all(
             [
-                "websocket" not in request.headers.get("Upgrade", "").lower(),
-                "upgrade" not in request.headers.get("Connection", "").lower(),
-                "Sec-WebSocket-Key" not in request.headers,
+                "websocket" in request.headers.get_directive("Upgrade", "").lower(),
+                "upgrade" in request.headers.get_directive("Connection", "").lower(),
+                "Sec-WebSocket-Key" in request.headers,
             ]
         ):
             raise ValueError("Request does not initiate websocket handshake")
 
     @staticmethod
     def _process_sec_websocket_key(request: Request) -> str:
-        key = request.headers.get("Sec-WebSocket-Key")
+        key = request.headers.get_directive("Sec-WebSocket-Key")
 
         if key is None:
             raise ValueError("Request does not have Sec-WebSocket-Key header")

adafruit_httpserver/server.py

@@ -243,7 +243,7 @@ def _receive_request(
 
         request = Request(self, sock, client_address, header_bytes)
 
-        content_length = int(request.headers.get("Content-Length", 0))
+        content_length = int(request.headers.get_directive("Content-Length", 0))
         received_body_bytes = request.body
 
         # Receiving remaining body bytes