Merge pull request #2311 from VWS-Python/beartype

Author: adamtheturtle

CHANGELOG.rst

@@ -6,6 +6,7 @@ Next
 
 * Breaking change: Exception names now end with ``Error``.
 * Use a timeout (30 seconds) when making requests to the VWS API.
+* Type hint changes: images are now ``io.BytesIO`` instances or ``io.BufferedRandom``.
 
 2024.02.19
 ------------

conftest.py

@@ -2,13 +2,25 @@
 
 from doctest import ELLIPSIS
 
+import pytest
+from beartype import beartype
 from sybil import Sybil
 from sybil.parsers.rest import (
     ClearNamespaceParser,
     DocTestParser,
     PythonCodeBlockParser,
 )
 
+
+def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
+    """
+    Apply the beartype decorator to all collected test functions.
+    """
+    for item in items:
+        if isinstance(item, pytest.Function):
+            item.obj = beartype(obj=item.obj)
+
+
 pytest_collect_file = Sybil(
     parsers=[
         ClearNamespaceParser(),

docs/source/conf.py

@@ -62,6 +62,7 @@
     "python": ("https://docs.python.org/3.12", None),
 }
 nitpicky = True
+nitpick_ignore = (("py:class", "BytesIO"), ("py:class", "BufferedRandom"))
 warning_is_error = True
 
 autoclass_content = "both"

src/vws/query.py

@@ -7,7 +7,7 @@
 import datetime
 import json
 from http import HTTPMethod, HTTPStatus
-from typing import Any, BinaryIO
+from typing import TYPE_CHECKING, Any
 from urllib.parse import urljoin
 
 import requests
@@ -29,8 +29,11 @@
 from vws.include_target_data import CloudRecoIncludeTargetData
 from vws.reports import QueryResult, TargetData
 
+if TYPE_CHECKING:
+    from io import BufferedRandom, BytesIO
 
-def _get_image_data(image: BinaryIO) -> bytes:
+
+def _get_image_data(image: BytesIO | BufferedRandom) -> bytes:
     """Get the data of an image file."""
     original_tell = image.tell()
     image.seek(0)
@@ -62,7 +65,7 @@ def __init__(
 
     def query(
         self,
-        image: BinaryIO,
+        image: BytesIO | BufferedRandom,
         max_num_results: int = 1,
         include_target_data: CloudRecoIncludeTargetData = (
             CloudRecoIncludeTargetData.TOP

src/vws/vws.py

@@ -9,7 +9,7 @@
 import time
 from datetime import date
 from http import HTTPMethod, HTTPStatus
-from typing import TYPE_CHECKING, BinaryIO
+from typing import TYPE_CHECKING
 from urllib.parse import urljoin
 
 import requests
@@ -51,9 +51,10 @@
 
 if TYPE_CHECKING:
     import io
+    from io import BufferedRandom, BytesIO
 
 
-def _get_image_data(image: BinaryIO) -> bytes:
+def _get_image_data(image: BytesIO | BufferedRandom) -> bytes:
     """Get the data of an image file."""
     original_tell = image.tell()
     image.seek(0)
@@ -238,7 +239,7 @@ def add_target(
         self,
         name: str,
         width: float,
-        image: BinaryIO,
+        image: BytesIO | BufferedRandom,
         application_metadata: str | None,
         *,
         active_flag: bool,
@@ -621,7 +622,7 @@ def update_target(
         target_id: str,
         name: str | None = None,
         width: float | None = None,
-        image: io.BytesIO | None = None,
+        image: io.BytesIO | io.BufferedRandom | None = None,
         active_flag: bool | None = None,
         application_metadata: str | None = None,
     ) -> None:

tests/conftest.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 import io
-from typing import TYPE_CHECKING, BinaryIO
+from typing import TYPE_CHECKING
 
 import pytest
 from mock_vws import MockVWS
@@ -68,7 +68,7 @@ def image_file(
 @pytest.fixture(params=["high_quality_image", "image_file"])
 def image(
     request: pytest.FixtureRequest,
-) -> BinaryIO:
+) -> io.BytesIO | io.BufferedRandom:
     """An image in any of the types that the API accepts."""
     result = request.getfixturevalue(request.param)
     assert isinstance(result, io.BytesIO | io.BufferedRandom)

tests/test_cloud_reco_exceptions.py

@@ -32,7 +32,7 @@
 
 def test_too_many_max_results(
     cloud_reco_client: CloudRecoService,
-    high_quality_image: io.BytesIO,
+    high_quality_image: io.BytesIO | io.BufferedRandom,
 ) -> None:
     """
     A ``MaxNumResultsOutOfRange`` error is raised if the given
@@ -53,7 +53,7 @@ def test_too_many_max_results(
 
 def test_image_too_large(
     cloud_reco_client: CloudRecoService,
-    png_too_large: io.BytesIO,
+    png_too_large: io.BytesIO | io.BufferedRandom,
 ) -> None:
     """
     A ``RequestEntityTooLarge`` exception is raised if an image which is too
@@ -82,7 +82,9 @@ def test_cloudrecoexception_inheritance() -> None:
         assert issubclass(subclass, CloudRecoError)
 
 
-def test_authentication_failure(high_quality_image: io.BytesIO) -> None:
+def test_authentication_failure(
+    high_quality_image: io.BytesIO | io.BufferedRandom,
+) -> None:
     """
     An ``AuthenticationFailure`` exception is raised when the client access key
     exists but the client secret key is incorrect.
@@ -101,7 +103,9 @@ def test_authentication_failure(high_quality_image: io.BytesIO) -> None:
         assert exc.value.response.status_code == HTTPStatus.UNAUTHORIZED
 
 
-def test_inactive_project(high_quality_image: io.BytesIO) -> None:
+def test_inactive_project(
+    high_quality_image: io.BytesIO | io.BufferedRandom,
+) -> None:
     """
     An ``InactiveProject`` exception is raised when querying an inactive
     database.

tests/test_query.py

@@ -25,7 +25,7 @@ class TestQuery:
     @staticmethod
     def test_no_matches(
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         An empty list is returned if there are no matches.
@@ -37,7 +37,7 @@ def test_no_matches(
     def test_match(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         Details of matching targets are returned.
@@ -60,7 +60,7 @@ class TestCustomBaseVWQURL:
     """
 
     @staticmethod
-    def test_custom_base_url(image: io.BytesIO) -> None:
+    def test_custom_base_url(image: io.BytesIO | io.BufferedRandom) -> None:
         """
         It is possible to use query a target to a database under a custom VWQ
         URL.
@@ -105,7 +105,7 @@ class TestMaxNumResults:
     def test_default(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         By default the maximum number of results is 1.
@@ -133,7 +133,7 @@ def test_default(
     def test_custom(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         It is possible to set a custom ``max_num_results``.
@@ -179,7 +179,7 @@ class TestIncludeTargetData:
     def test_default(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         By default, target data is only returned in the top match.
@@ -211,7 +211,7 @@ def test_default(
     def test_top(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         When ``CloudRecoIncludeTargetData.TOP`` is given, target data is only
@@ -245,7 +245,7 @@ def test_top(
     def test_none(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         When ``CloudRecoIncludeTargetData.NONE`` is given, target data is not
@@ -279,7 +279,7 @@ def test_none(
     def test_all(
         vws_client: VWS,
         cloud_reco_client: CloudRecoService,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         When ``CloudRecoIncludeTargetData.ALL`` is given, target data is

tests/test_vws.py

@@ -8,7 +8,7 @@
 import datetime
 import secrets
 import uuid
-from typing import TYPE_CHECKING, BinaryIO
+from typing import TYPE_CHECKING
 
 import pytest
 from freezegun import freeze_time
@@ -38,7 +38,7 @@ class TestAddTarget:
     @pytest.mark.parametrize("active_flag", [True, False])
     def test_add_target(
         vws_client: VWS,
-        image: BinaryIO,
+        image: io.BytesIO | io.BufferedRandom,
         application_metadata: bytes | None,
         cloud_reco_client: CloudRecoService,
         *,
@@ -82,7 +82,7 @@ def test_add_target(
     @staticmethod
     def test_add_two_targets(
         vws_client: VWS,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         No exception is raised when adding two targets with different names.
@@ -105,7 +105,7 @@ class TestCustomBaseVWSURL:
     """
 
     @staticmethod
-    def test_custom_base_url(image: io.BytesIO) -> None:
+    def test_custom_base_url(image: io.BytesIO | io.BufferedRandom) -> None:
         """
         It is possible to use add a target to a database under a custom VWS
         URL.
@@ -137,7 +137,7 @@ class TestListTargets:
     @staticmethod
     def test_list_targets(
         vws_client: VWS,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         It is possible to get a list of target IDs.
@@ -167,7 +167,7 @@ class TestDelete:
     @staticmethod
     def test_delete_target(
         vws_client: VWS,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         It is possible to delete a target.
@@ -194,7 +194,7 @@ class TestGetTargetSummaryReport:
     @staticmethod
     def test_get_target_summary_report(
         vws_client: VWS,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         Details of a target are returned by ``get_target_summary_report``.
@@ -263,7 +263,7 @@ class TestGetTargetRecord:
     @staticmethod
     def test_get_target_record(
         vws_client: VWS,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         Details of a target are returned by ``get_target_record``.
@@ -298,7 +298,7 @@ class TestWaitForTargetProcessed:
     @staticmethod
     def test_wait_for_target_processed(
         vws_client: VWS,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         It is possible to wait until a target is processed.
@@ -318,7 +318,7 @@ def test_wait_for_target_processed(
 
     @staticmethod
     def test_default_seconds_between_requests(
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         By default, 0.2 seconds are waited between polling requests.
@@ -370,7 +370,7 @@ def test_default_seconds_between_requests(
 
     @staticmethod
     def test_custom_seconds_between_requests(
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         It is possible to customize the time waited between polling requests.
@@ -421,7 +421,7 @@ def test_custom_seconds_between_requests(
             assert report.request_usage == expected_requests
 
     @staticmethod
-    def test_custom_timeout(image: io.BytesIO) -> None:
+    def test_custom_timeout(image: io.BytesIO | io.BufferedRandom) -> None:
         """
         It is possible to set a maximum timeout.
         """
@@ -465,7 +465,7 @@ class TestGetDuplicateTargets:
     @staticmethod
     def test_get_duplicate_targets(
         vws_client: VWS,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         It is possible to get the IDs of similar targets.
@@ -499,8 +499,8 @@ class TestUpdateTarget:
     @staticmethod
     def test_update_target(
         vws_client: VWS,
-        image: io.BytesIO,
-        different_high_quality_image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
+        different_high_quality_image: io.BytesIO | io.BufferedRandom,
         cloud_reco_client: CloudRecoService,
     ) -> None:
         """
@@ -558,7 +558,7 @@ def test_update_target(
     @staticmethod
     def test_no_fields_given(
         vws_client: VWS,
-        image: io.BytesIO,
+        image: io.BytesIO | io.BufferedRandom,
     ) -> None:
         """
         It is possible to give no update fields.