Allow the public methods to take file-like objects, not just io.BytesIO

Author: adamtheturtle

CHANGELOG.rst

@@ -4,6 +4,8 @@ Changelog
 Next
 ----
 
+* Support file-like objects in every method which accepts a file.
+
 2023.03.05
 ------------
 

README.md

@@ -51,7 +51,6 @@ shutil.copy(image, new_image)
 <!--pytest-codeblocks:cont-->
 
 ```python
-import io
 import pathlib
 
 from vws import VWS, CloudRecoService
@@ -73,17 +72,15 @@ name = 'my_image_name'
 
 image = pathlib.Path('high_quality_image.jpg')
 with image.open(mode='rb') as my_image_file:
-   my_image = io.BytesIO(my_image_file.read())
-
-target_id = vws_client.add_target(
-    name=name,
-    width=1,
-    image=my_image,
-    active_flag=True,
-    application_metadata=None,
-)
-vws_client.wait_for_target_processed(target_id=target_id)
-matching_targets = cloud_reco_client.query(image=my_image)
+    target_id = vws_client.add_target(
+        name=name,
+        width=1,
+        image=my_image_file,
+        active_flag=True,
+        application_metadata=None,
+    )
+    vws_client.wait_for_target_processed(target_id=target_id)
+    matching_targets = cloud_reco_client.query(image=my_image_file)
 
 assert matching_targets[0].target_id == target_id
 ```

docs/source/index.rst

@@ -43,7 +43,6 @@ See the :doc:`api-reference` for full usage details.
 
 .. testcode:: simple
 
-   import io
    import pathlib
 
    from vws import VWS, CloudRecoService
@@ -65,17 +64,15 @@ See the :doc:`api-reference` for full usage details.
 
    image = pathlib.Path('high_quality_image.jpg')
    with image.open(mode='rb') as my_image_file:
-      my_image = io.BytesIO(my_image_file.read())
-
-   target_id = vws_client.add_target(
-       name=name,
-       width=1,
-       image=my_image,
-       active_flag=True,
-       application_metadata=None,
-   )
-   vws_client.wait_for_target_processed(target_id=target_id)
-   matching_targets = cloud_reco_client.query(image=my_image)
+      target_id = vws_client.add_target(
+         name=name,
+         width=1,
+         image=my_image_file,
+         active_flag=True,
+         application_metadata=None,
+      )
+      vws_client.wait_for_target_processed(target_id=target_id)
+      matching_targets = cloud_reco_client.query(image=my_image_file)
 
    assert matching_targets[0].target_id == target_id
 
@@ -109,7 +106,6 @@ To write unit tests for code which uses this library, without using your Vuforia
 
 .. testcode:: testing
 
-   import io
    import pathlib
 
    from mock_vws.database import VuforiaDatabase
@@ -131,15 +127,13 @@ To write unit tests for code which uses this library, without using your Vuforia
 
        image = pathlib.Path('high_quality_image.jpg')
        with image.open(mode='rb') as my_image_file:
-           my_image = io.BytesIO(my_image_file.read())
-
-       target_id = vws_client.add_target(
-           name="example_image_name",
-           width=1,
-           image=my_image,
-           application_metadata=None,
-           active_flag=True,
-       )
+         target_id = vws_client.add_target(
+            name="example_image_name",
+            width=1,
+            image=my_image_file,
+            application_metadata=None,
+            active_flag=True,
+         )
 
 .. testcleanup:: testing
 

src/vws/query.py

@@ -6,7 +6,7 @@
 
 import datetime
 from http import HTTPStatus
-from typing import TYPE_CHECKING
+from typing import BinaryIO
 from urllib.parse import urljoin
 
 import requests
@@ -27,8 +27,13 @@
 from vws.include_target_data import CloudRecoIncludeTargetData
 from vws.reports import QueryResult, TargetData
 
-if TYPE_CHECKING:
-    import io
+
+def _get_image_data(image: BinaryIO) -> bytes:
+    original_tell = image.tell()
+    image.seek(0)
+    image_data = image.read()
+    image.seek(original_tell)
+    return image_data
 
 
 class CloudRecoService:
@@ -54,7 +59,7 @@ def __init__(
 
     def query(
         self,
-        image: io.BytesIO,
+        image: BinaryIO,
         max_num_results: int = 1,
         include_target_data: CloudRecoIncludeTargetData = (
             CloudRecoIncludeTargetData.TOP
@@ -96,7 +101,7 @@ def query(
         Returns:
             An ordered list of target details of matching targets.
         """
-        image_content = image.getvalue()
+        image_content = _get_image_data(image=image)
         body = {
             "image": ("image.jpeg", image_content, "image/jpeg"),
             "max_num_results": (None, int(max_num_results), "text/plain"),

src/vws/vws.py

@@ -8,7 +8,7 @@
 import json
 from datetime import date
 from time import sleep
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, BinaryIO
 from urllib.parse import urljoin
 
 import requests
@@ -51,6 +51,14 @@
     import io
 
 
+def _get_image_data(image: BinaryIO) -> bytes:
+    original_tell = image.tell()
+    image.seek(0)
+    image_data = image.read()
+    image.seek(original_tell)
+    return image_data
+
+
 def _target_api_request(
     server_access_key: str,
     server_secret_key: str,
@@ -204,7 +212,7 @@ def add_target(
         self,
         name: str,
         width: int | float,
-        image: io.BytesIO,
+        image: BinaryIO,
         application_metadata: str | None,
         *,
         active_flag: bool,
@@ -255,7 +263,7 @@ def add_target(
                 occurred". This has been seen to happen when the given name
                 includes a bad character.
         """
-        image_data = image.getvalue()
+        image_data = _get_image_data(image=image)
         image_data_encoded = base64.b64encode(image_data).decode("ascii")
 
         data = {
@@ -644,7 +652,7 @@ def update_target(
             data["width"] = width
 
         if image is not None:
-            image_data = image.getvalue()
+            image_data = _get_image_data(image=image)
             image_data_encoded = base64.b64encode(image_data).decode("ascii")
             data["image"] = image_data_encoded
 

tests/conftest.py

@@ -4,19 +4,21 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import io
+from typing import TYPE_CHECKING, BinaryIO
 
 import pytest
 from mock_vws import MockVWS
 from mock_vws.database import VuforiaDatabase
 from vws import VWS, CloudRecoService
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator
+    from collections.abc import Generator
+    from pathlib import Path
 
 
 @pytest.fixture(name="_mock_database")
-def mock_database() -> Iterator[VuforiaDatabase]:
+def mock_database() -> Generator[VuforiaDatabase, None, None]:
     """
     Yield a mock ``VuforiaDatabase``.
     """
@@ -29,7 +31,7 @@ def mock_database() -> Iterator[VuforiaDatabase]:
 @pytest.fixture()
 def vws_client(_mock_database: VuforiaDatabase) -> VWS:
     """
-    Yield a VWS client which connects to a mock database.
+    A VWS client which connects to a mock database.
     """
     return VWS(
         server_access_key=_mock_database.server_access_key,
@@ -40,9 +42,33 @@ def vws_client(_mock_database: VuforiaDatabase) -> VWS:
 @pytest.fixture()
 def cloud_reco_client(_mock_database: VuforiaDatabase) -> CloudRecoService:
     """
-    Yield a ``CloudRecoService`` client which connects to a mock database.
+    A ``CloudRecoService`` client which connects to a mock database.
     """
     return CloudRecoService(
         client_access_key=_mock_database.client_access_key,
         client_secret_key=_mock_database.client_secret_key,
     )
+
+
+@pytest.fixture()
+def image_file(
+    high_quality_image: io.BytesIO,
+    tmp_path: Path,
+) -> Generator[io.BufferedRandom, None, None]:
+    """An image file object."""
+    file = tmp_path / "image.jpg"
+    file.touch()
+    with file.open("r+b") as fileobj:
+        buffer = high_quality_image.getvalue()
+        fileobj.write(buffer)
+        yield fileobj
+
+
+@pytest.fixture(params=["high_quality_image", "image_file"])
+def image(
+    request: pytest.FixtureRequest,
+) -> BinaryIO:
+    """An image in any of the types that the API accepts."""
+    result = request.getfixturevalue(request.param)
+    assert isinstance(result, io.BytesIO | io.BufferedRandom)
+    return result