From 7c42fb04f8db90ea8b72311819aa39a0b2423a35 Mon Sep 17 00:00:00 2001
From: Heikki Henriksen <heikkih@gmail.com>
Date: Mon, 27 Apr 2026 17:56:47 +0200
Subject: [PATCH] Add get_transfer() and cancel_transfer() for file transfer
 monitoring

Exposes GET /api/v1/transfer as get_transfer() -> Transfer | None,
returning None on 204 (no active transfer). Also exposes
DELETE /api/v1/transfer/{id} as cancel_transfer(transfer_id).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 pyprusalink/__init__.py | 21 ++++++++++++++++++++-
 pyprusalink/types.py    | 16 ++++++++++++++++
 tests/test_prusalink.py | 36 ++++++++++++++++++++++++++++++++++++
 3 files changed, 72 insertions(+), 1 deletion(-)

diff --git a/pyprusalink/__init__.py b/pyprusalink/__init__.py
index 6cc332f..dcaa6f1 100644
--- a/pyprusalink/__init__.py
+++ b/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']"""
diff --git a/pyprusalink/types.py b/pyprusalink/types.py
index 83556b7..0c3fa6a 100644
--- a/pyprusalink/types.py
+++ b/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]
diff --git a/tests/test_prusalink.py b/tests/test_prusalink.py
index 37e14d0..501ad4d 100644
--- a/tests/test_prusalink.py
+++ b/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(