Add support for bytes formatted images to submit_image and submit_gallery

AUTHORS.rst

@@ -81,4 +81,5 @@ Source Contributors
 - Josh Kim `@jsk56143 <https://github.com/jsk56143>`_
 - Rolf Campbell `@endlisnis <https://github.com/endlisnis>`_
 - zacc `@zacc <https://github.com/zacc>`_
+- Connor Colabella `@redowul <https://github.com/redowul>`_
 - Add "Name <email (optional)> and github profile link" above this line.

praw/models/reddit/subreddit.py

@@ -4,9 +4,9 @@
 import socket
 from copy import deepcopy
 from csv import writer
-from io import StringIO
+from io import BytesIO, StringIO
 from json import dumps, loads
-from os.path import basename, dirname, isfile, join
+from os.path import basename, isfile
 from typing import TYPE_CHECKING, Any, Dict, Generator, Iterator, List, Optional, Union
 from urllib.parse import urljoin
 from warnings import warn
@@ -209,13 +209,22 @@ def _subreddit_list(*, other_subreddits, subreddit):
 
     @staticmethod
     def _validate_gallery(images):
-        for image in images:
-            image_path = image.get("image_path", "")
-            if image_path:
-                if not isfile(image_path):
-                    raise TypeError(f"{image_path!r} is not a valid image path.")
+        for index, image in enumerate(images):
+            image_path = image.get("image_path")
+            image_fp = image.get("image_fp")
+            if image_path is not None and image_fp is None:
+                if isinstance(image_path, str):
+                    if not isfile(image_path):
+                        raise ValueError(f"{image_path!r} is not a valid file path.")
+            elif image_path is None and image_fp is not None:
+                if not isinstance(image_fp, bytes):
+                    raise TypeError(
+                        f"'image_fp' dictionary value at index {index} contains an invalid bytes object."
+                    )  # do not log bytes value, it is long and not human readable
             else:
-                raise TypeError("'image_path' is required.")
+                raise TypeError(
+                    f"Values for keys image_path and image_fp are null for dictionary at index {index}."
+                )
             if not len(image.get("caption", "")) <= 180:
                 raise TypeError("Caption must be 180 characters or less.")
 
@@ -643,10 +652,15 @@ def _submit_media(
         url = ws_update["payload"]["redirect"]
         return self._reddit.submission(url=url)
 
-    def _read_and_post_media(self, media_path, upload_url, upload_data):
-        with open(media_path, "rb") as media:
+    def _read_and_post_media(self, media_path, media_fp, upload_url, upload_data):
+        if media_path is not None and media_fp is None:
+            with open(media_path, "rb") as media:
+                response = self._reddit._core._requestor._http.post(
+                    upload_url, data=upload_data, files={"file": media}
+                )
+        elif media_path is None and media_fp is not None:
             response = self._reddit._core._requestor._http.post(
-                upload_url, data=upload_data, files={"file": media}
+                upload_url, data=upload_data, files={"file": BytesIO(media_fp)}
             )
         return response
 
@@ -655,6 +669,7 @@ def _upload_media(
         *,
         expected_mime_prefix: Optional[str] = None,
         media_path: str,
+        media_fp: bytes,
         upload_type: str = "link",
     ):
         """Upload media and return its URL and a websocket (Undocumented endpoint).
@@ -669,23 +684,86 @@ def _upload_media(
             finished, or it can be ignored.
 
         """
-        if media_path is None:
-            media_path = join(
-                dirname(dirname(dirname(__file__))), "images", "PRAW logo.png"
-            )
-
-        file_name = basename(media_path).lower()
-        file_extension = file_name.rpartition(".")[2]
         mime_type = {
             "png": "image/png",
             "mov": "video/quicktime",
             "mp4": "video/mp4",
             "jpg": "image/jpeg",
             "jpeg": "image/jpeg",
             "gif": "image/gif",
-        }.get(
-            file_extension, "image/jpeg"
-        )  # default to JPEG
+        }
+        if media_path is not None and media_fp is None:
+            if isfile(media_path):
+                file_name = basename(media_path).lower()
+                file_extension = file_name.rpartition(".")[2]
+                mime_type = mime_type.get(
+                    file_extension, "image/jpeg"
+                )  # default to JPEG
+            else:
+                raise TypeError("media_path does not reference a file.")
+        elif media_path is None and media_fp is not None:
+            if isinstance(media_fp, bytes):
+                magic_number = [
+                    int(aByte) for aByte in media_fp[:8]
+                ]  # gets the format indicator
+                file_headers = {
+                    tuple(
+                        [
+                            int(aByte)
+                            for aByte in bytes(
+                                [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]
+                            )
+                        ]
+                    ): "png",
+                    tuple(
+                        [int(aByte) for aByte in bytes([0x6D, 0x6F, 0x6F, 0x76])]
+                    ): "mov",
+                    tuple(
+                        [
+                            int(aByte)
+                            for aByte in bytes(
+                                [0x66, 0x74, 0x79, 0x70, 0x69, 0x73, 0x6F, 0x6D]
+                            )
+                        ]
+                    ): "mp4",
+                    tuple(
+                        [int(aByte) for aByte in bytes([0xFF, 0xD8, 0xFF, 0xE0])]
+                    ): "jpg",
+                    tuple(
+                        [
+                            int(aByte)
+                            for aByte in bytes(
+                                [0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10, 0x4A, 0x46]
+                            )
+                        ]
+                    ): "jpeg",
+                    tuple(
+                        [
+                            int(aByte)
+                            for aByte in bytes([0x47, 0x49, 0x46, 0x38, 0x37, 0x61])
+                        ]
+                    ): "gif",
+                }
+                for size in range(4, 10, 2):  # size will equal 4, 6, 8
+                    file_extension = file_headers.get(tuple(magic_number[:size]))
+                    if file_extension is not None:
+                        mime_type = mime_type.get(
+                            file_extension, "image/jpeg"
+                        )  # default to JPEG
+                        file_name = (
+                            mime_type.split("/")[0] + "." + mime_type.split("/")[1]
+                        )
+                        break
+                if file_extension is None:
+                    raise TypeError(
+                        "media_fp does not represent an accepted file format"
+                        " (png, mov, mp4, jpg, jpeg, gif.)"
+                    )
+            else:
+                raise TypeError("media_fp is not of type bytes.")
+        else:
+            raise TypeError("media_path and media_fp are null.")
+
         if (
             expected_mime_prefix is not None
             and mime_type.partition("/")[0] != expected_mime_prefix
@@ -703,7 +781,9 @@ def _upload_media(
         upload_url = f"https:{upload_lease['action']}"
         upload_data = {item["name"]: item["value"] for item in upload_lease["fields"]}
 
-        response = self._read_and_post_media(media_path, upload_url, upload_data)
+        response = self._read_and_post_media(
+            media_path, media_fp, upload_url, upload_data
+        )
         if not response.ok:
             self._parse_xml_response(response)
         try:
@@ -1040,7 +1120,7 @@ def submit(
     def submit_gallery(
         self,
         title: str,
-        images: List[Dict[str, str]],
+        images: List[Dict[str, str]] = None,
         *,
         collection_id: Optional[str] = None,
         discussion_type: Optional[str] = None,
@@ -1053,9 +1133,11 @@ def submit_gallery(
         """Add an image gallery submission to the subreddit.
 
         :param title: The title of the submission.
-        :param images: The images to post in dict with the following structure:
-            ``{"image_path": "path", "caption": "caption", "outbound_url": "url"}``,
-            only ``image_path`` is required.
+        :param images: The images to post in dict with one of the following two
+            structures: ``{"image_path": "path", "caption": "caption", "outbound_url":
+            "url"}`` and ``{"image_fp": "file_pointer", "caption": "caption",
+            "outbound_url": "url"}``, only ``image_path`` and ``image_fp`` are required
+            for each given structure.
         :param collection_id: The UUID of a :class:`.Collection` to add the
             newly-submitted post to.
         :param discussion_type: Set to ``"CHAT"`` to enable live discussion instead of
@@ -1132,7 +1214,8 @@ def submit_gallery(
                     "outbound_url": image.get("outbound_url", ""),
                     "media_id": self._upload_media(
                         expected_mime_prefix="image",
-                        media_path=image["image_path"],
+                        media_path=image.get("image_path"),
+                        media_fp=image.get("image_fp"),
                         upload_type="gallery",
                     )[0],
                 }
@@ -1162,8 +1245,9 @@ def submit_gallery(
     def submit_image(
         self,
         title: str,
-        image_path: str,
         *,
+        image_path: Optional[str] = None,
+        image_fp: Optional[bytes] = None,
         collection_id: Optional[str] = None,
         discussion_type: Optional[str] = None,
         flair_id: Optional[str] = None,
@@ -1185,7 +1269,9 @@ def submit_image(
         :param flair_text: If the template's ``flair_text_editable`` value is ``True``,
             this value will set a custom text (default: ``None``). ``flair_id`` is
             required when ``flair_text`` is provided.
-        :param image_path: The path to an image, to upload and post.
+        :param image_path: The path to an image, to upload and post. (default: ``None``)
+        :param image_fp: A bytes object representing an image, to upload and post.
+            (default: ``None``)
         :param nsfw: Whether the submission should be marked NSFW (default: ``False``).
         :param resubmit: When ``False``, an error will occur if the URL has already been
             submitted (default: ``True``).
@@ -1255,8 +1341,9 @@ def submit_image(
                 data[key] = value
 
         image_url, websocket_url = self._upload_media(
-            expected_mime_prefix="image", media_path=image_path
+            expected_mime_prefix="image", media_path=image_path, media_fp=image_fp
         )
+
         data.update(kind="image", url=image_url)
         if without_websockets:
             websocket_url = None

tests/unit/models/reddit/test_subreddit.py

@@ -144,7 +144,7 @@ def test_submit_failure(self):
             subreddit.submit("Cool title", selftext="", url="b")
         assert str(excinfo.value) == message
 
-    def test_submit_gallery__missing_path(self):
+    def test_submit_gallery__missing_image_path(self):
         message = "'image_path' is required."
         subreddit = Subreddit(self.reddit, display_name="name")
 
@@ -154,14 +154,12 @@ def test_submit_gallery__missing_path(self):
             )
         assert str(excinfo.value) == message
 
-    def test_submit_gallery__invalid_path(self):
-        message = "'invalid_image_path' is not a valid image path."
+    def test_submit_gallery__invalid_image_path(self):
+        message = "'invalid_image' is not a valid file path."
         subreddit = Subreddit(self.reddit, display_name="name")
 
         with pytest.raises(TypeError) as excinfo:
-            subreddit.submit_gallery(
-                "Cool title", [{"image_path": "invalid_image_path"}]
-            )
+            subreddit.submit_gallery("Cool title", [{"image_path": "invalid_image"}])
         assert str(excinfo.value) == message
 
     def test_submit_gallery__too_long_caption(self):