diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1bb2cf2..a7269c3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -29,6 +29,9 @@ jobs: - name: Test with unittest run: | python -m unittest thttp.py + env: + MEDIAPUB_URL: ${{ secrets.MEDIAPUB_URL }} + MEDIAPUB_TOKEN: ${{ secrets.MEDIAPUB_TOKEN }} black: runs-on: ubuntu-latest @@ -128,3 +131,6 @@ jobs: run: | coverage run -m unittest thttp.py coverage report -m --fail-under=95 + env: + MEDIAPUB_URL: ${{ secrets.MEDIAPUB_URL }} + MEDIAPUB_TOKEN: ${{ secrets.MEDIAPUB_TOKEN }} diff --git a/README.md b/README.md index 99a682e..cf47d0e 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ It's features include: - Loading JSON from the response - Returning error responses instead of throwing exceptions from `urllib` - `pretty()` function for printing responses +- Ability to upload files using the `{"file": open("file.png", "rb")}` style Future features: @@ -25,7 +26,7 @@ Future features: - Improve handling of non-utf-8 requests - Improve handling of non-utf-8 responses -_Note: this project is not intended to solve all use cases that can be achieved with urllib, requests or other HTTP libraries. The intent is to provide a lightweight tool that simplifies some of the most common use cases for developers._ +_Note: this project is not intended to solve all use cases that can be achieved with urllib, requests, httpx, or other HTTP libraries. The intent is to provide a lightweight tool that simplifies some of the most common use cases for developers._ ## Installation diff --git a/test-image.png b/test-image.png new file mode 100644 index 0000000..45ff853 Binary files /dev/null and b/test-image.png differ diff --git a/thttp.py b/thttp.py index b2606a5..8cfabc5 100644 --- a/thttp.py +++ b/thttp.py @@ -7,6 +7,8 @@ import gzip import json as json_lib +import mimetypes +import secrets import ssl from base64 import b64encode from collections import namedtuple @@ -42,6 +44,7 @@ def request( cookiejar=None, basic_auth=None, timeout=None, + files={}, # note: experimental ): """ Returns a (named)tuple with the following properties: @@ -66,6 +69,9 @@ def request( if method not in ["POST", "PATCH", "PUT"] and (json or data): raise Exception("Request method must POST, PATCH or PUT if json or data is provided") + if files and method != "POST": + raise Exception("Request method must be POST when uploading files") + if not timeout: timeout = 60 @@ -76,6 +82,31 @@ def request( data = urlencode(data).encode() elif isinstance(data, str): data = data.encode() + elif files: + boundary = secrets.token_hex() + + headers["Content-Type"] = f"multipart/form-data; boundary={boundary}" + data = b"" + + for key, file in files.items(): + file_data = file.read() # okay, we want this to stay as a byte-string + if isinstance(file_data, str): + file_data = file_data.encode("utf-8") + fn = file.name + + mime, _ = mimetypes.guess_type(fn) + if not mime: + print("Using default mimetype") + mime = "application/octet-stream" + + data += b"--" + boundary.encode() + b"\r\n" + data += b'Content-Disposition: form-data; name="' + key.encode() + b'"; filename="' + fn.encode() + b'"\r\n' + data += b"Content-Type: " + mime.encode() + b"\r\n\r\n" + data += file_data + b"\r\n" + data += b"--" + boundary.encode() + b"--\r\n" + + data = data + headers["Content-Length"] = len(data) if basic_auth and len(basic_auth) == 2 and "authorization" not in headers: username, password = basic_auth @@ -155,6 +186,7 @@ def pretty(response, headers_only=False): import contextlib # noqa: E402 +import os # noqa: E402 import unittest # noqa: E402 from io import StringIO # noqa: E402 from unittest.mock import patch # noqa: E402 @@ -303,3 +335,22 @@ def test_thttp_with_mocked_response(self): with patch("thttp.request", side_effect=[mocked_response]): response = request("https://example.org") self.assertEqual("mocked", response.json["response"]) + + def test_upload_single_file(self): + token = os.environ.get("MEDIAPUB_TOKEN") + url = os.environ.get("MEDIAPUB_URL") + + if not token or not url: + self.skipTest("Skipping media upload test because environment variables are not available") + + for fn in ["test-image.png", "LICENSE.md"]: + with open(fn, "rb" if fn.endswith("png") else "r") as f: + response = request( + url, + headers={"Authorization": f"Bearer {token}"}, + files={"file": f}, + method="POST", + ) + + self.assertEqual(response.status, 201) + self.assertTrue("location" in response.headers)