review comments

Author: ajha-ss

singlestoredb/management/files.py

@@ -424,13 +424,36 @@ def is_dir(self, path: PathLike) -> bool:
     def is_file(self, path: PathLike) -> bool:
         pass
 
+    @overload
+    @abstractmethod
+    def listdir(
+        self,
+        path: PathLike = '/',
+        *,
+        recursive: bool = False,
+        return_objects: Literal[True],
+    ) -> List[FilesObject]:
+        pass
+
+    @overload
     @abstractmethod
     def listdir(
         self,
         path: PathLike = '/',
         *,
         recursive: bool = False,
-    ) -> Union[List[str], List[Union[str, Dict[str, Any]]]]:
+        return_objects: Literal[False] = False,
+    ) -> List[str]:
+        pass
+
+    @abstractmethod
+    def listdir(
+        self,
+        path: PathLike = '/',
+        *,
+        recursive: bool = False,
+        return_objects: bool = False,
+    ) -> Union[List[str], List[FilesObject]]:
         pass
 
     @abstractmethod
@@ -914,46 +937,44 @@ def is_file(self, path: PathLike) -> bool:
     def _listdir(
         self, path: PathLike, *,
         recursive: bool = False,
-        return_meta: bool = False,
-    ) -> List[Union[str, Dict[str, Any]]]:
+        return_objects: bool = False,
+    ) -> List[Union[str, FilesObject]]:
         """
-        Return the names (or metadata) of files in a directory.
+        Return the names (or FilesObject instances) of files in a directory.
 
         Parameters
         ----------
         path : Path or str
             Path to the folder
         recursive : bool, optional
             Should folders be listed recursively?
-        return_meta : bool, optional
-            If True, return list of dicts with 'path' and 'type'. Otherwise just paths.
+        return_objects : bool, optional
+            If True, return list of FilesObject instances. Otherwise just paths.
         """
         res = self._manager._get(
             f'files/fs/{self._location}/{path}',
         ).json()
 
         if recursive:
-            out: List[Union[str, Dict[str, Any]]] = []
+            out: List[Union[str, FilesObject]] = []
             for item in res.get('content') or []:
-                if return_meta:
-                    out.append(
-                        {'path': item['path'], 'type': item['type']},
-                    )
+                if return_objects:
+                    out.append(FilesObject.from_dict(item, self))
                 else:
                     out.append(item['path'])
                 if item['type'] == 'directory':
                     out.extend(
                         self._listdir(
                             item['path'],
                             recursive=recursive,
-                            return_meta=return_meta,
+                            return_objects=return_objects,
                         ),
                     )
             return out
 
-        if return_meta:
+        if return_objects:
             return [
-                {'path': x['path'], 'type': x['type']}
+                FilesObject.from_dict(x, self)
                 for x in (res.get('content') or [])
             ]
         return [x['path'] for x in (res.get('content') or [])]
@@ -964,8 +985,8 @@ def listdir(
         path: PathLike = '/',
         *,
         recursive: bool = False,
-        return_meta: Literal[True] = ...,
-    ) -> List[Dict[str, Any]]:
+        return_objects: Literal[True],
+    ) -> List[FilesObject]:
         ...
 
     @overload
@@ -974,7 +995,7 @@ def listdir(
         path: PathLike = '/',
         *,
         recursive: bool = False,
-        return_meta: Literal[False] = ...,
+        return_objects: Literal[False] = False,
     ) -> List[str]:
         ...
 
@@ -983,8 +1004,8 @@ def listdir(
         path: PathLike = '/',
         *,
         recursive: bool = False,
-        return_meta: bool = False,
-    ) -> Union[List[str], List[Dict[str, Any]]]:
+        return_objects: bool = False,
+    ) -> Union[List[str], List[FilesObject]]:
         """
         List the files / folders at the given path.
 
@@ -993,39 +1014,39 @@ def listdir(
         path : Path or str, optional
             Path to the file location
 
-        return_meta : bool, optional
-            If True, return list of dicts with 'path' and 'type'. Otherwise just paths.
+        return_objects : bool, optional
+            If True, return list of FilesObject instances. Otherwise just paths.
 
         Returns
         -------
-        List[str] or List[dict]
+        List[str] or List[FilesObject]
 
         """
         path = re.sub(r'^(\./|/)+', r'', str(path))
         path = re.sub(r'/+$', r'', path) + '/'
 
         # Validate via listing GET; if response lacks 'content', it's not a directory
         try:
-            out = self._listdir(path, recursive=recursive, return_meta=return_meta)
+            out = self._listdir(path, recursive=recursive, return_objects=return_objects)
         except Exception as exc:
             # If the path doesn't exist or isn't a directory, _listdir will fail
             raise NotADirectoryError(f'path is not a directory: {path}') from exc
 
         if path != '/':
             path_n = len(path.split('/')) - 1
-            if return_meta:
-                result: List[Dict[str, Any]] = []
+            if return_objects:
+                result: List[FilesObject] = []
                 for item in out:
-                    if isinstance(item, dict):
-                        rel = '/'.join(item['path'].split('/')[path_n:])
-                        item['path'] = rel
+                    if isinstance(item, FilesObject):
+                        rel = '/'.join(item.path.split('/')[path_n:])
+                        item.path = rel
                         result.append(item)
                 return result
             return ['/'.join(str(x).split('/')[path_n:]) for x in out]
 
-        # _listdir guarantees homogeneous type based on return_meta
-        if return_meta:
-            return cast(List[Dict[str, Any]], out)
+        # _listdir guarantees homogeneous type based on return_objects
+        if return_objects:
+            return cast(List[FilesObject], out)
         return cast(List[str], out)
 
     def download_file(
@@ -1139,13 +1160,13 @@ def download_folder(
             raise OSError('target path already exists; use overwrite=True to replace')
 
         # listdir validates directory; no extra info call needed
-        entries = self.listdir(path, recursive=True, return_meta=True)
+        entries = self.listdir(path, recursive=True, return_objects=True)
         for entry in entries:
-            # Each entry is a dict with path relative to root and type
-            if not isinstance(entry, dict):  # defensive: skip unexpected
+            # Each entry is a FilesObject with path relative to root and type
+            if not isinstance(entry, FilesObject):  # defensive: skip unexpected
                 continue
-            rel_path = entry['path']
-            if entry['type'] == 'directory':
+            rel_path = entry.path
+            if entry.type == 'directory':
                 # Ensure local directory exists; no remote call needed
                 target_dir = os.path.normpath(os.path.join(local_path, rel_path))
                 os.makedirs(target_dir, exist_ok=True)

singlestoredb/management/workspace.py

@@ -10,9 +10,12 @@
 import time
 from collections.abc import Mapping
 from typing import Any
+from typing import cast
 from typing import Dict
 from typing import List
+from typing import Literal
 from typing import Optional
+from typing import overload
 from typing import Union
 
 from .. import config
@@ -447,58 +450,117 @@ def is_file(self, stage_path: PathLike) -> bool:
                 return False
             raise
 
-    def _listdir(self, stage_path: PathLike, *, recursive: bool = False) -> List[str]:
+    def _listdir(
+        self, stage_path: PathLike, *,
+        recursive: bool = False,
+        return_objects: bool = False,
+    ) -> List[Union[str, 'FilesObject']]:
         """
-        Return the names of files in a directory.
+        Return the names (or FilesObject instances) of files in a directory.
 
         Parameters
         ----------
         stage_path : Path or str
             Path to the folder in Stage
         recursive : bool, optional
             Should folders be listed recursively?
+        return_objects : bool, optional
+            If True, return list of FilesObject instances. Otherwise just paths.
 
         """
+        from .files import FilesObject
         res = self._manager._get(
             re.sub(r'/+$', r'/', f'stage/{self._deployment_id}/fs/{stage_path}'),
         ).json()
         if recursive:
-            out = []
+            out: List[Union[str, FilesObject]] = []
             for item in res['content'] or []:
-                out.append(item['path'])
+                if return_objects:
+                    out.append(FilesObject.from_dict(item, self))
+                else:
+                    out.append(item['path'])
                 if item['type'] == 'directory':
-                    out.extend(self._listdir(item['path'], recursive=recursive))
+                    out.extend(
+                        self._listdir(
+                            item['path'],
+                            recursive=recursive,
+                            return_objects=return_objects,
+                        ),
+                    )
             return out
+        if return_objects:
+            return [
+                FilesObject.from_dict(x, self)
+                for x in res['content'] or []
+            ]
         return [x['path'] for x in res['content'] or []]
 
+    @overload
+    def listdir(
+        self,
+        stage_path: PathLike = '/',
+        *,
+        recursive: bool = False,
+        return_objects: Literal[True],
+    ) -> List['FilesObject']:
+        ...
+
+    @overload
     def listdir(
         self,
         stage_path: PathLike = '/',
         *,
         recursive: bool = False,
+        return_objects: Literal[False] = False,
     ) -> List[str]:
+        ...
+
+    def listdir(
+        self,
+        stage_path: PathLike = '/',
+        *,
+        recursive: bool = False,
+        return_objects: bool = False,
+    ) -> Union[List[str], List['FilesObject']]:
         """
         List the files / folders at the given path.
 
         Parameters
         ----------
         stage_path : Path or str, optional
             Path to the stage location
+        return_objects : bool, optional
+            If True, return list of FilesObject instances. Otherwise just paths.
 
         Returns
         -------
-        List[str]
+        List[str] or List[FilesObject]
 
         """
+        from .files import FilesObject
         stage_path = re.sub(r'^(\./|/)+', r'', str(stage_path))
         stage_path = re.sub(r'/+$', r'', stage_path) + '/'
 
         if self.is_dir(stage_path):
-            out = self._listdir(stage_path, recursive=recursive)
+            out = self._listdir(
+                stage_path,
+                recursive=recursive,
+                return_objects=return_objects,
+            )
             if stage_path != '/':
                 stage_path_n = len(stage_path.split('/')) - 1
-                out = ['/'.join(x.split('/')[stage_path_n:]) for x in out]
-            return out
+                if return_objects:
+                    result: List[FilesObject] = []
+                    for item in out:
+                        if isinstance(item, FilesObject):
+                            rel = '/'.join(item.path.split('/')[stage_path_n:])
+                            item.path = rel
+                            result.append(item)
+                    return result
+                out = ['/'.join(str(x).split('/')[stage_path_n:]) for x in out]
+            if return_objects:
+                return cast(List[FilesObject], out)
+            return cast(List[str], out)
 
         raise NotADirectoryError(f'stage path is not a directory: {stage_path}')
 
@@ -577,7 +639,7 @@ def download_folder(
         if not self.is_dir(stage_path):
             raise NotADirectoryError(f'stage path is not a directory: {stage_path}')
 
-        for f in self.listdir(stage_path, recursive=True):
+        for f in self.listdir(stage_path, recursive=True, return_objects=False):
             if self.is_dir(f):
                 continue
             target = os.path.normpath(os.path.join(local_path, f))