migrate folder to use cluster folder apis where relevant

runhouse/resources/folders/folder.py

@@ -142,8 +142,14 @@ def mv(self, system, path: Optional[str] = None) -> None:
  if path is None:
  raise ValueError("A destination path must be specified.")
 
- dest_path = Path(path).expanduser()
- src_path = Path(self.path).expanduser()
+ src_path = Path(self.path)
+ dest_path = Path(path)
+
+ if isinstance(self.system, Cluster):
+ return self.system._mv(path=src_path, dest_path=dest_path)
+
+ dest_path = dest_path.expanduser()
+ src_path = src_path.expanduser()
 
  if not src_path.exists():
  raise FileNotFoundError(f"The source path {src_path} does not exist.")
@@ -287,10 +293,15 @@ def _to_data_store(
 
  def mkdir(self):
  """Create the folder in specified file system if it doesn't already exist."""
- path = Path(self.path).expanduser()
- if not path.exists():
- path.mkdir(parents=True, exist_ok=True)
- logger.info(f"Folder created in path: {path}")
+ path = Path(self.path)
+ if isinstance(self.system, Cluster):
+ self.system._mkdir(path)
+ else:
+ path = path.expanduser()
+ if not path.exists():
+ path.mkdir(parents=True, exist_ok=True)
+
+ logger.info(f"Folder created in path: {self.path}")
 
  def _to_cluster(self, dest_cluster, path=None):
  """Copy the folder from a file or cluster source onto a destination cluster."""
@@ -498,7 +509,11 @@ def ls(self, full_paths: bool = True, sort: bool = False) -> List:
  sort (Optional[bool]): Whether to sort the folder contents by time modified.
  Defaults to ``False``.
  """
- paths = [p for p in Path(self.path).expanduser().iterdir()]
+ path = Path(self.path)
+ if isinstance(self.system, Cluster):
+ return self.system._ls(path)
+
+ paths = [p for p in path.expanduser().iterdir()]
 
  # Sort the paths by modification time if sort is True
  if sort:
@@ -518,7 +533,13 @@ def resources(self, full_paths: bool = False):
  """
  try:
  resources = [
- path for path in self.ls() if (Path(path) / "config.json").exists()
+ path
+ for path in (
+ self.system._ls(path=Path(path))
+ if isinstance(self.system, Cluster)
+ else self.ls()
+ )
+ if (Path(path) / "config.json").exists()
  ]
  except FileNotFoundError:
  return []
@@ -653,6 +674,11 @@ def get(self, name, mode="rb", encoding=None):
  Example:
  >>> contents = my_folder.get(file_name)
  """
+ if isinstance(self.system, Cluster):
+ return self.system._get(
+ path=Path(self.path) / name, mode=mode, encoding=encoding
+ )
+
  with self.open(name, mode=mode, encoding=encoding) as f:
  return f.read()
 
@@ -682,8 +708,11 @@ def rm(self, contents: list = None, recursive: bool = True):
  Example:
  >>> my_folder.rm()
  """
- folder_path = Path(self.path).expanduser()
+ path = Path(self.path)
+ if isinstance(self.system, Cluster):
+ return self.system._rm(path, contents, recursive)
 
+ folder_path = path.expanduser()
  if contents:
  for content in contents:
  content_path = folder_path / content
@@ -725,15 +754,19 @@ def put(self, contents, overwrite=False, mode: str = "wb"):
  Example:
  >>> my_folder.put(contents={"filename.txt": data})
  """
- self.mkdir()
 
- full_path = str(Path(self.path).expanduser())
  # Handle lists of resources just for convenience
  if isinstance(contents, list):
  for resource in contents:
  self.put(resource, overwrite=overwrite)
  return
 
+ if isinstance(self.system, Cluster):
+ return self.system._put(
+ path=Path(self.path), contents=contents, overwrite=overwrite, mode=mode
+ )
+
+ full_path = str(Path(self.path).expanduser())
  if isinstance(contents, Folder):
  if contents.path is None: # Should only be the case when Folder is created
  contents.path = os.path.join(full_path, contents.name)

runhouse/servers/http/http_client.py

@@ -21,14 +21,15 @@
 from runhouse.servers.http.http_utils import (
  CallParams,
  DeleteObjectParams,
+ FolderOperation,
  FolderParams,
  GetObjectParams,
  handle_response,
  OutputType,
  PutObjectParams,
  PutResourceParams,
  RenameObjectParams,
- serialize_data, FolderOperation,
+ serialize_data,
 )