Simplify upload file

Author: patrick91

src/fastapi_cloud_cli/commands/deploy.py

@@ -7,7 +7,7 @@
 from itertools import cycle
 from pathlib import Path, PurePosixPath
 from textwrap import dedent
-from typing import Annotated, Any
+from typing import Annotated, Any, BinaryIO, cast
 
 import fastar
 import rignore
@@ -247,7 +247,7 @@ def progress_callback(bytes_read: int) -> None:
             upload_response = client.post(
                 upload_data.url,
                 data=upload_data.fields,
-                files={"file": archive_file_with_progress},
+                files={"file": cast(BinaryIO, archive_file_with_progress)},
             )
 
         upload_response.raise_for_status()

src/fastapi_cloud_cli/utils/progress_file.py

@@ -1,10 +1,10 @@
-from collections.abc import Callable, Iterator
+from collections.abc import Callable
 from datetime import datetime
 from typing import Any, BinaryIO
 
 
-class ProgressFile(BinaryIO):
-    """Wraps a file object to track read progress."""
+class ProgressFile:
+    """Wrap a binary file object and report upload progress as it is read."""
 
     def __init__(
         self,
@@ -26,70 +26,5 @@ def read(self, n: int = -1) -> bytes:
             self._last_update_time = now_
         return data
 
-    def fileno(self) -> int:
-        return self._file.fileno()
-
-    @property
-    def name(self) -> str:
-        return self._file.name
-
-    def seek(self, offset: int, whence: int = 0) -> int:
-        return self._file.seek(offset, whence)
-
-    def tell(self) -> int:
-        return self._file.tell()
-
-    def __iter__(self) -> Iterator[bytes]:
-        return self._file.__iter__()
-
-    def __next__(self) -> bytes:
-        return next(self._file)  # pragma: no cover
-
-    @property
-    def mode(self) -> str:
-        return self._file.mode  # pragma: no cover
-
-    def readable(self) -> bool:
-        return self._file.readable()  # pragma: no cover
-
-    def seekable(self) -> bool:
-        return self._file.seekable()  # pragma: no cover
-
-    # Methods below are just to satisfy the BinaryIO interface
-
-    def write(self, *_args: Any, **kwargs: Any) -> int:
-        raise NotImplementedError()  # pragma: no cover
-
-    def readline(self, *_args: Any, **kwargs: Any) -> bytes:
-        raise NotImplementedError()  # pragma: no cover
-
-    def readlines(self, *_args: Any, **kwargs: Any) -> list[bytes]:
-        raise NotImplementedError()  # pragma: no cover
-
-    def writelines(self, *_args: Any, **kwargs: Any) -> None:
-        raise NotImplementedError()  # pragma: no cover
-
-    def __enter__(self) -> BinaryIO:
-        raise NotImplementedError()  # pragma: no cover
-
-    def __exit__(self, *_args: Any) -> None:
-        raise NotImplementedError()  # pragma: no cover
-
-    def close(self) -> None:
-        raise NotImplementedError()  # pragma: no cover
-
-    @property
-    def closed(self) -> bool:
-        raise NotImplementedError()  # pragma: no cover
-
-    def flush(self) -> None:
-        raise NotImplementedError()  # pragma: no cover
-
-    def isatty(self) -> bool:
-        raise NotImplementedError()  # pragma: no cover
-
-    def writable(self) -> bool:
-        raise NotImplementedError()  # pragma: no cover
-
-    def truncate(self, *_args: Any, **kwargs: Any) -> int:
-        raise NotImplementedError()  # pragma: no cover
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._file, name)

tests/test_progress_file.py

@@ -15,13 +15,6 @@ def _make_file(
     return f
 
 
-def test_read_returns_data() -> None:
-    file = _make_file(b"abc")
-    pf = ProgressFile(file, progress_callback=lambda _: None)
-
-    assert pf.read() == b"abc"
-
-
 def test_read_with_size() -> None:
     file = _make_file(b"abcdef")
     pf = ProgressFile(file, progress_callback=lambda _: None)
@@ -94,18 +87,7 @@ def test_name_property() -> None:
     assert pf.name == "test.tar.gz"
 
 
-def test_seek_and_tell() -> None:
-    file = _make_file(b"abcdef")
-    pf = ProgressFile(file, progress_callback=lambda _: None)
-
-    pf.seek(3)
-    assert pf.tell() == 3
-
-    pf.seek(0)
-    assert pf.tell() == 0
-
-
-def test_seek_resets_bytes_read() -> None:
+def test_callback_uses_current_file_position_after_seek() -> None:
     file = _make_file(b"abcde")
     mock_callback = Mock()
 
@@ -120,11 +102,3 @@ def test_seek_resets_bytes_read() -> None:
     pf.read(3)
 
     mock_callback.assert_has_calls([call(3), call(5)])
-
-
-def test_iter_delegates() -> None:
-    file = _make_file(b"line1\nline2\n")
-    pf = ProgressFile(file, progress_callback=lambda _: None)
-
-    lines = list(pf)
-    assert lines == [b"line1\n", b"line2\n"]