add TransferClient.operation_stat

changelog.d/20240301_143526_aaschaer_stat.rst

@@ -0,0 +1,4 @@
+ Added
+ ~~~~~
+
+ - Added `TransferClient,operation_stat` helper method for getting the status of a path on a collection (:pr:`NUMBER`)

src/globus_sdk/_testing/data/transfer/operation_stat.py

@@ -0,0 +1,27 @@
+from globus_sdk._testing.models import RegisteredResponse, ResponseSet
+
+from ._common import ENDPOINT_ID
+
+RESPONSES = ResponseSet(
+    metadata={"endpoint_id": ENDPOINT_ID},
+    default=RegisteredResponse(
+        service="transfer",
+        method="GET",
+        path=f"/operation/endpoint/{ENDPOINT_ID}/stat",
+        json={
+            "DATA_TYPE": "file",
+            "group": "tutorial",
+            "last_modified": "2023-12-18 16:52:50+00:00",
+            "link_group": None,
+            "link_last_modified": None,
+            "link_size": None,
+            "link_target": None,
+            "link_user": None,
+            "name": "file1.txt",
+            "permissions": "0644",
+            "size": 4,
+            "type": "file",
+            "user": "tutorial",
+        },
+    ),
+)

src/globus_sdk/services/transfer/client.py

@@ -1235,6 +1235,40 @@ def operation_rename(
             query_params=query_params,
         )
 
+    def operation_stat(
+        self,
+        endpoint_id: UUIDLike,
+        path: str | None = None,
+        *,
+        local_user: str | None = None,
+        query_params: dict[str, t.Any] | None = None,
+    ) -> response.GlobusHTTPResponse:
+        """
+        :param endpoint_id: The ID of the collection on which to do the stat operation
+        :param path: Path on the collection to do the stat operation on
+        :param local_user: Optional value passed to identity mapping specifying which
+            local user account to map to. Only usable with Globus Connect Server v5
+            mapped collections.
+        :param query_params: Additional passthrough query parameters
+            .. tab-item:: API Info
+
+                ``GET /operation/endpoint/<endpoint_id>/stat``
+
+                .. extdoclink:: Get File or Directory Status
+                    :ref: transfer/file_operations/#stat
+        """
+        if query_params is None:
+            query_params = {}
+        if path is not None:
+            query_params["path"] = path
+        if local_user is not None:
+            query_params["local_user"] = local_user
+
+        log.info(f"TransferClient.operation_stat({endpoint_id}, {query_params})")
+        return self.get(
+            f"operation/endpoint/{endpoint_id}/stat", query_params=query_params
+        )
+
     def operation_symlink(
         self,
         endpoint_id: UUIDLike,

tests/functional/services/transfer/test_operation_stat.py

@@ -0,0 +1,22 @@
+"""
+Tests for TransferClient.operation_stat
+"""
+
+import urllib.parse
+
+from globus_sdk._testing import get_last_request, load_response
+
+
+def test_operation_stat(client):
+    meta = load_response(client.operation_stat).metadata
+    endpoint_id = meta["endpoint_id"]
+    path = "/home/share/godata/file1.txt"
+    res = client.operation_stat(endpoint_id, path)
+
+    assert res["name"] == "file1.txt"
+    assert res["type"] == "file"
+    assert res["size"] == 4
+
+    req = get_last_request()
+    parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query)
+    assert parsed_qs == {"path": [path]}