Add get_transfer() and cancel_transfer() for file transfer monitoring

pyprusalink/__init__.py

@@ -4,7 +4,14 @@
 
 from httpx import AsyncClient
 from pyprusalink.client import ApiClient
-from pyprusalink.types import JobInfo, PrinterInfo, PrinterStatus, Storage, VersionInfo
+from pyprusalink.types import (
+    JobInfo,
+    PrinterInfo,
+    PrinterStatus,
+    Storage,
+    Transfer,
+    VersionInfo,
+)
 from pyprusalink.types_legacy import LegacyPrinterStatus
 
 
@@ -76,6 +83,18 @@ async def get_storage(self) -> list[Storage]:
         async with self.client.request("GET", "/api/v1/storage") as response:
             return response.json()["storage_list"]
 
+    async def get_transfer(self) -> Transfer | None:
+        """Get active transfer. Returns None when no transfer is in progress."""
+        async with self.client.request("GET", "/api/v1/transfer") as response:
+            if response.status_code == 204:
+                return None
+            return response.json()
+
+    async def cancel_transfer(self, transfer_id: int) -> None:
+        """Cancel the transfer with the given id."""
+        async with self.client.request("DELETE", f"/api/v1/transfer/{transfer_id}"):
+            pass
+
     # Prusa Link Web UI still uses the old endpoints and it seems that the new v1 endpoint doesn't support this yet
     async def get_file(self, path: str) -> bytes:
         """Get a files such as Thumbnails or Icons. Path comes from the current job['file']['refs']['thumbnail']"""

pyprusalink/types.py

@@ -170,3 +170,19 @@ class Storage(TypedDict):
     total_space: NotRequired[int]
     print_files: NotRequired[int]
     system_files: NotRequired[int]
+
+
+class Transfer(TypedDict):
+    """An active file transfer returned by /api/v1/transfer."""
+
+    type: str
+    display_name: str
+    path: str
+    progress: float
+    transferred: int
+    time_transferring: int
+    to_print: bool
+    id: NotRequired[int]
+    url: NotRequired[str]
+    size: NotRequired[str]
+    time_remaining: NotRequired[int]

tests/test_prusalink.py

@@ -221,6 +221,42 @@ async def test_get_storage(pl, respx_mock):
     assert result[0]["available"] is True
 
 
+async def test_get_transfer(pl, respx_mock):
+    respx_mock.get(f"{HOST}/api/v1/transfer").mock(
+        return_value=httpx.Response(
+            200,
+            json={
+                "id": 7,
+                "type": "FROM_WEB",
+                "display_name": "model.gcode",
+                "path": "/usb",
+                "size": "2393142",
+                "progress": 42.25,
+                "transferred": 1011000,
+                "time_remaining": 61,
+                "time_transferring": 42,
+                "to_print": False,
+            },
+        )
+    )
+    result = await pl.get_transfer()
+    assert result["type"] == "FROM_WEB"
+    assert result["progress"] == 42.25
+
+
+async def test_get_transfer_none(pl, respx_mock):
+    """Returns None when no transfer is active."""
+    respx_mock.get(f"{HOST}/api/v1/transfer").mock(return_value=httpx.Response(204))
+    assert await pl.get_transfer() is None
+
+
+async def test_cancel_transfer(pl, respx_mock):
+    respx_mock.delete(f"{HOST}/api/v1/transfer/7").mock(
+        return_value=httpx.Response(204)
+    )
+    await pl.cancel_transfer(7)
+
+
 async def test_get_file(pl, respx_mock):
     thumbnail_bytes = b"\x89PNG\r\nfake-image-data"
     respx_mock.get(f"{HOST}/api/thumbnails/test.png").mock(