Core: Add touch API

tosfs/core.py

@@ -15,7 +15,7 @@
 """The core module of TOSFS."""
 import logging
 import os
-from typing import List, Optional, Tuple, Union
+from typing import Any, List, Optional, Tuple, Union
 
 import tos
 from fsspec import AbstractFileSystem
@@ -231,6 +231,50 @@ def rmdir(self, path: str) -> None:
         except Exception as e:
             raise TosfsError(f"Tosfs failed with unknown error: {e}") from e
 
+    def touch(self, path: str, truncate: bool = True, **kwargs: Any) -> None:
+        """Create an empty file at the given path.
+
+        Parameters
+        ----------
+        path : str
+            The path of the file to create.
+        truncate : bool, optional
+            Whether to truncate the file if it already exists (default is True).
+        **kwargs : Any, optional
+            Additional arguments.
+
+        Raises
+        ------
+        FileExistsError
+            If the file already exists and `truncate` is False.
+        TosfsError
+            If there is an unknown error while creating the file.
+        tos.exceptions.TosClientError
+            If there is a client error while creating the file.
+        tos.exceptions.TosServerError
+            If there is a server error while creating the file.
+
+        Examples
+        --------
+        >>> fs = TosFileSystem()
+        >>> fs.touch("tos://mybucket/myfile")
+
+        """
+        path = self._strip_protocol(path)
+        bucket, key, _ = self._split_path(path)
+
+        if not truncate and self.exists(path):
+            raise FileExistsError(f"File {path} already exists.")
+
+        try:
+            self.tos_client.put_object(bucket, key)
+        except tos.exceptions.TosClientError as e:
+            raise e
+        except tos.exceptions.TosServerError as e:
+            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/conftest.py

@@ -56,6 +56,6 @@ def temporary_workspace(
     yield workspace
     try:
         tosfs.rmdir(f"{bucket}/{workspace}/")
-    except Exception:
-        logger.error("Ignore exception.")
+    except Exception as e:
+        logger.error(f"Ignore exception : {e}.")
     assert not tosfs.exists(f"{bucket}/{workspace}/")

tosfs/tests/test_tosfs.py

@@ -109,3 +109,27 @@ def test_rmdir(tosfs: TosFileSystem, bucket: str, temporary_workspace: str) -> N
     assert f"{bucket}/{temporary_workspace}" not in tosfs.ls(
         bucket, detail=False, refresh=True
     )
+
+
+def test_touch(tosfs: TosFileSystem, bucket: str, temporary_workspace: str) -> None:
+    file_name = random_path()
+    assert not tosfs.exists(f"{bucket}/{temporary_workspace}/{file_name}")
+    tosfs.touch(f"{bucket}/{temporary_workspace}/{file_name}")
+    assert tosfs.exists(f"{bucket}/{temporary_workspace}/{file_name}")
+    assert tosfs.info(f"{bucket}/{temporary_workspace}/{file_name}")["size"] == 0
+
+    with pytest.raises(FileExistsError):
+        tosfs.touch(f"{bucket}/{temporary_workspace}/{file_name}", truncate=False)
+
+    tosfs.rm_file(f"{bucket}/{temporary_workspace}/{file_name}")
+    tosfs.touch(f"{bucket}/{temporary_workspace}/{file_name}", truncate=False)
+    assert tosfs.exists(f"{bucket}/{temporary_workspace}/{file_name}")
+
+    tosfs.rm_file(f"{bucket}/{temporary_workspace}/{file_name}")
+    tosfs.tos_client.put_object(
+        bucket=bucket, key=f"{temporary_workspace}/{file_name}", content="hello world"
+    )
+    assert tosfs.info(f"{bucket}/{temporary_workspace}/{file_name}")["size"] > 0
+    tosfs.touch(f"{bucket}/{temporary_workspace}/{file_name}", truncate=True)
+    assert tosfs.info(f"{bucket}/{temporary_workspace}/{file_name}")["size"] == 0
+    tosfs.rm_file(f"{bucket}/{temporary_workspace}/{file_name}")