Files arg

adafruit_requests.py

@@ -41,6 +41,7 @@
 
 import errno
 import json as json_module
+import random
 import sys
 
 from adafruit_connection_manager import get_connection_manager
@@ -343,6 +344,14 @@ def iter_content(self, chunk_size: int = 1, decode_unicode: bool = False) -> byt
         self.close()
 
 
+def _generate_boundary_str():
+    hex_characters = "0123456789abcdef"
+    _boundary = ""
+    for _ in range(32):
+        _boundary += random.choice(hex_characters)
+    return _boundary
+
+
 class Session:
     """HTTP session that shares sockets and ssl context."""
 
@@ -405,8 +414,7 @@ def _send_header(self, socket, header, value):
             self._send_as_bytes(socket, value)
         self._send(socket, b"\r\n")
 
-    # pylint: disable=too-many-arguments
-    def _send_request(
+    def _send_request(  # pylint: disable=too-many-arguments
         self,
         socket: SocketType,
         host: str,
@@ -415,7 +423,8 @@ def _send_request(
         headers: Dict[str, str],
         data: Any,
         json: Any,
-    ):
+        files: Optional[Dict[str, tuple]],
+    ):  # pylint: disable=too-many-branches,too-many-locals,too-many-statements
         # Check headers
         self._check_headers(headers)
 
@@ -425,6 +434,7 @@ def _send_request(
         # If json is sent, set content type header and convert to string
         if json is not None:
             assert data is None
+            assert files is None
             content_type_header = "application/json"
             data = json_module.dumps(json)
 
@@ -441,13 +451,70 @@ def _send_request(
         if data and isinstance(data, str):
             data = bytes(data, "utf-8")
 
+        if data is None:
+            data = b""
+
         self._send_as_bytes(socket, method)
         self._send(socket, b" /")
         self._send_as_bytes(socket, path)
         self._send(socket, b" HTTP/1.1\r\n")
 
         # create lower-case supplied header list
         supplied_headers = {header.lower() for header in headers}
+        boundary_str = None
+
+        # pylint: disable=too-many-nested-blocks
+        if files is not None and isinstance(files, dict):
+            boundary_str = _generate_boundary_str()
+            content_type_header = f"multipart/form-data; boundary={boundary_str}"
+
+            for fieldname in files.keys():
+                if not fieldname.endswith("-name"):
+                    if files[fieldname][0] is not None:
+                        file_content = files[fieldname][1].read()
+
+                        data += b"--" + boundary_str.encode() + b"\r\n"
+                        data += (
+                            b'Content-Disposition: form-data; name="'
+                            + fieldname.encode()
+                            + b'"; filename="'
+                            + files[fieldname][0].encode()
+                            + b'"\r\n'
+                        )
+                        if len(files[fieldname]) >= 3:
+                            data += (
+                                b"Content-Type: "
+                                + files[fieldname][2].encode()
+                                + b"\r\n"
+                            )
+                        if len(files[fieldname]) >= 4:
+                            for custom_header_key in files[fieldname][3].keys():
+                                data += (
+                                    custom_header_key.encode()
+                                    + b": "
+                                    + files[fieldname][3][custom_header_key].encode()
+                                    + b"\r\n"
+                                )
+                        data += b"\r\n"
+                        data += file_content + b"\r\n"
+                    else:
+                        # filename is None
+                        data += b"--" + boundary_str.encode() + b"\r\n"
+                        data += (
+                            b'Content-Disposition: form-data; name="'
+                            + fieldname.encode()
+                            + b'"; \r\n'
+                        )
+                        if len(files[fieldname]) >= 3:
+                            data += (
+                                b"Content-Type: "
+                                + files[fieldname][2].encode()
+                                + b"\r\n"
+                            )
+                        data += b"\r\n"
+                        data += files[fieldname][1].encode() + b"\r\n"
+
+            data += b"--" + boundary_str.encode() + b"--"
 
         # Send headers
         if not "host" in supplied_headers:
@@ -478,6 +545,7 @@ def request(
         stream: bool = False,
         timeout: float = 60,
         allow_redirects: bool = True,
+        files: Optional[Dict[str, tuple]] = None,
     ) -> Response:
         """Perform an HTTP request to the given url which we will parse to determine
         whether to use SSL ('https://') or not. We can also send some provided 'data'
@@ -526,7 +594,9 @@ def request(
             )
             ok = True
             try:
-                self._send_request(socket, host, method, path, headers, data, json)
+                self._send_request(
+                    socket, host, method, path, headers, data, json, files
+                )
             except OSError as exc:
                 last_exc = exc
                 ok = False

examples/wifi/expanded/requests_wifi_file_upload.py

@@ -0,0 +1,27 @@
+# SPDX-FileCopyrightText: 2024 Tim Cocks for Adafruit Industries
+# SPDX-License-Identifier: MIT
+
+import adafruit_connection_manager
+import wifi
+
+import adafruit_requests
+
+URL = "https://httpbin.org/post"
+
+pool = adafruit_connection_manager.get_radio_socketpool(wifi.radio)
+ssl_context = adafruit_connection_manager.get_radio_ssl_context(wifi.radio)
+requests = adafruit_requests.Session(pool, ssl_context)
+
+with open("raspi_snip.png", "rb") as file_handle:
+    files = {
+        "file": (
+            "raspi_snip.png",
+            file_handle,
+            "image/png",
+            {"CustomHeader": "BlinkaRocks"},
+        ),
+        "othervalue": (None, "HelloWorld"),
+    }
+
+    with requests.post(URL, files=files) as resp:
+        print(resp.content)

examples/wifi/expanded/requests_wifi_file_upload_image.png.license

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: 2024 Tim Cocks
+# SPDX-License-Identifier: CC-BY-4.0