Core: Implement isfile API

tosfs/core.py

@@ -323,6 +323,37 @@ def isdir(self, path: str) -> bool:
         except Exception as e:
             raise TosfsError(f"Tosfs failed with unknown error: {e}") from e
 
+    def isfile(self, path: str) -> bool:
+        """Check if the path is a file.
+
+        Parameters
+        ----------
+        path : str
+            The path to check.
+
+        Returns
+        -------
+        bool
+            True if the path is a file, False otherwise.
+
+        """
+        if path.endswith("/"):
+            return False
+
+        bucket, key, _ = self._split_path(path)
+        try:
+            # Attempt to get the object metadata
+            self.tos_client.head_object(bucket, key)
+            return True
+        except tos.exceptions.TosClientError as e:
+            raise e
+        except tos.exceptions.TosServerError as e:
+            if e.status_code == SERVER_RESPONSE_CODE_NOT_FOUND:
+                return False
+            raise e
+        except Exception as e:
+            raise TosfsError(f"Tosfs failed with unknown error: {e}") from e
+
     def _bucket_info(self, bucket: str) -> dict:
         """Get the information of a bucket.
 

tosfs/tests/test_tosfs.py

@@ -149,4 +149,14 @@ def test_isdir(tosfs: TosFileSystem, bucket: str, temporary_workspace: str) -> N
     assert not tosfs.isdir(f"{bucket}/{temporary_workspace}/{file_name}")
     assert not tosfs.isdir(f"{bucket}/{temporary_workspace}/{file_name}/")
 
+
+def test_isfile(tosfs: TosFileSystem, bucket: str, temporary_workspace: str) -> None:
+    file_name = random_path()
+    tosfs.tos_client.put_object(bucket=bucket, key=f"{temporary_workspace}/{file_name}")
+    assert tosfs.isfile(f"{bucket}/{temporary_workspace}/{file_name}")
+    assert not tosfs.isfile(f"{bucket}/{temporary_workspace}/{file_name}/")
+    assert not tosfs.isfile(f"{bucket}/{temporary_workspace}/nonexistfile")
+    assert not tosfs.isfile(f"{bucket}/{temporary_workspace}")
+    assert not tosfs.isfile(f"{bucket}/{temporary_workspace}/")
+
     tosfs._rm(f"{bucket}/{temporary_workspace}/{file_name}")