From d84f8f9b4b1b9bc6aa461fb118fbfe41917b32de Mon Sep 17 00:00:00 2001
From: Aditya Jha <ajha@singlestore.com>
Date: Thu, 4 Dec 2025 11:52:38 +0530
Subject: [PATCH 1/9] optimize management api calls

---
 singlestoredb/management/files.py | 69 +++++++++++++++++++++----------
 1 file changed, 48 insertions(+), 21 deletions(-)

diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py
index ac3561ad7..91a2455a8 100644
--- a/singlestoredb/management/files.py
+++ b/singlestoredb/management/files.py
@@ -908,9 +908,9 @@ def is_file(self, path: PathLike) -> bool:
                 return False
             raise
 
-    def _listdir(self, path: PathLike, *, recursive: bool = False) -> List[str]:
+    def _listdir(self, path: PathLike, *, recursive: bool = False, return_meta: bool = False) -> List[Union[str, Dict[str, Any]]]:
         """
-        Return the names of files in a directory.
+        Return the names (or metadata) of files in a directory.
 
         Parameters
         ----------
@@ -918,28 +918,35 @@ def _listdir(self, path: PathLike, *, recursive: bool = False) -> List[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.
         """
         res = self._manager._get(
             f'files/fs/{self._location}/{path}',
         ).json()
 
         if recursive:
-            out = []
-            for item in res['content'] or []:
-                out.append(item['path'])
+            out: List[Union[str, Dict[str, Any]]] = []
+            for item in res.get('content') or []:
+                if return_meta:
+                    out.append({'path': item['path'], 'type': item['type']})
+                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_meta=return_meta))
             return out
 
-        return [x['path'] for x in res['content'] or []]
+        if return_meta:
+            return [{'path': x['path'], 'type': x['type']} for x in (res.get('content') or [])]
+        return [x['path'] for x in (res.get('content') or [])]
 
     def listdir(
         self,
         path: PathLike = '/',
         *,
         recursive: bool = False,
-    ) -> List[str]:
+        return_meta: bool = False,
+    ) -> List[Union[str, Dict[str, Any]]]:
         """
         List the files / folders at the given path.
 
@@ -948,21 +955,32 @@ 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.
+
         Returns
         -------
-        List[str]
+        List[str] or List[dict]
 
         """
         path = re.sub(r'^(\./|/)+', r'', str(path))
         path = re.sub(r'/+$', r'', path) + '/'
 
-        if not self.is_dir(path):
+        # Single validation GET (info) rather than is_dir + info later
+        info = self.info(path)
+        if info.type != 'directory':
             raise NotADirectoryError(f'path is not a directory: {path}')
 
-        out = self._listdir(path, recursive=recursive)
+        out = self._listdir(path, recursive=recursive, return_meta=return_meta)
         if path != '/':
             path_n = len(path.split('/')) - 1
-            out = ['/'.join(x.split('/')[path_n:]) for x in out]
+            if return_meta:
+                for i in range(len(out)):
+                    if isinstance(out[i], dict):
+                        rel = '/'.join(out[i]['path'].split('/')[path_n:])
+                        out[i]['path'] = rel
+            else:
+                out = ['/'.join(str(x).split('/')[path_n:]) for x in out]
         return out
 
     def download_file(
@@ -1036,17 +1054,26 @@ def download_folder(
         if local_path is not None and not overwrite and os.path.exists(local_path):
             raise OSError('target path already exists; use overwrite=True to replace')
 
-        if not self.is_dir(path):
+        # Validate directory with single info call
+        info = self.info(path)
+        if info.type != 'directory':
             raise NotADirectoryError(f'path is not a directory: {path}')
 
-        files = self.listdir(path, recursive=True)
-        for f in files:
-            remote_path = os.path.join(path, f)
-            if self.is_dir(remote_path):
+        entries = self.listdir(path, recursive=True, return_meta=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
+                continue
+            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)
                 continue
-            target = os.path.normpath(os.path.join(local_path, f))
-            os.makedirs(os.path.dirname(target), exist_ok=True)
-            self.download_file(remote_path, target, overwrite=overwrite)
+            remote_path = os.path.join(path, rel_path)
+            target_file = os.path.normpath(os.path.join(local_path, rel_path))
+            os.makedirs(os.path.dirname(target_file), exist_ok=True)
+            self.download_file(remote_path, target_file, overwrite=overwrite)
 
     def remove(self, path: PathLike) -> None:
         """

From 70870ffa785b496dcff5eb58f3ee07ed23c6e675 Mon Sep 17 00:00:00 2001
From: Aditya Jha <ajha@singlestore.com>
Date: Thu, 4 Dec 2025 14:51:15 +0530
Subject: [PATCH 2/9] remove duplicate calls

---
 singlestoredb/management/files.py | 23 ++++++++++-------------
 1 file changed, 10 insertions(+), 13 deletions(-)

diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py
index 91a2455a8..1556fc50e 100644
--- a/singlestoredb/management/files.py
+++ b/singlestoredb/management/files.py
@@ -966,12 +966,12 @@ def listdir(
         path = re.sub(r'^(\./|/)+', r'', str(path))
         path = re.sub(r'/+$', r'', path) + '/'
 
-        # Single validation GET (info) rather than is_dir + info later
-        info = self.info(path)
-        if info.type != 'directory':
-            raise NotADirectoryError(f'path is not a directory: {path}')
-
-        out = self._listdir(path, recursive=recursive, return_meta=return_meta)
+        # Validate via listing GET; if response lacks 'content', it's not a directory
+        try:
+            out = self._listdir(path, recursive=recursive, return_meta=return_meta)
+        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:
@@ -990,6 +990,7 @@ def download_file(
         *,
         overwrite: bool = False,
         encoding: Optional[str] = None,
+        _skip_dir_check: bool = False,
     ) -> Optional[Union[bytes, str]]:
         """
         Download the content of a file path.
@@ -1013,7 +1014,7 @@ def download_file(
         """
         if local_path is not None and not overwrite and os.path.exists(local_path):
             raise OSError('target file already exists; use overwrite=True to replace')
-        if self.is_dir(path):
+        if not _skip_dir_check and self.is_dir(path):
             raise IsADirectoryError(f'file path is a directory: {path}')
 
         out = self._manager._get(
@@ -1054,11 +1055,7 @@ def download_folder(
         if local_path is not None and not overwrite and os.path.exists(local_path):
             raise OSError('target path already exists; use overwrite=True to replace')
 
-        # Validate directory with single info call
-        info = self.info(path)
-        if info.type != 'directory':
-            raise NotADirectoryError(f'path is not a directory: {path}')
-
+        # listdir validates directory; no extra info call needed
         entries = self.listdir(path, recursive=True, return_meta=True)
         for entry in entries:
             # Each entry is a dict with path relative to root and type
@@ -1073,7 +1070,7 @@ def download_folder(
             remote_path = os.path.join(path, rel_path)
             target_file = os.path.normpath(os.path.join(local_path, rel_path))
             os.makedirs(os.path.dirname(target_file), exist_ok=True)
-            self.download_file(remote_path, target_file, overwrite=overwrite)
+            self.download_file(remote_path, target_file, overwrite=overwrite, _skip_dir_check=True)
 
     def remove(self, path: PathLike) -> None:
         """

From 618a7b0af5d3aba2ff75c7f3ce997ebc6036a164 Mon Sep 17 00:00:00 2001
From: Aditya Jha <ajha@singlestore.com>
Date: Thu, 18 Dec 2025 11:45:19 +0530
Subject: [PATCH 3/9] review comment

---
 singlestoredb/management/files.py | 42 +++++++++++++++++++++++++++++--
 1 file changed, 40 insertions(+), 2 deletions(-)

diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py
index 1556fc50e..e62851fde 100644
--- a/singlestoredb/management/files.py
+++ b/singlestoredb/management/files.py
@@ -990,7 +990,6 @@ def download_file(
         *,
         overwrite: bool = False,
         encoding: Optional[str] = None,
-        _skip_dir_check: bool = False,
     ) -> Optional[Union[bytes, str]]:
         """
         Download the content of a file path.
@@ -1011,6 +1010,45 @@ def download_file(
         bytes or str - ``local_path`` is None
         None - ``local_path`` is a Path or str
 
+        """
+        return self._download_file(
+            path,
+            local_path=local_path,
+            overwrite=overwrite,
+            encoding=encoding,
+            _skip_dir_check=False,
+        )
+
+    def _download_file(
+        self,
+        path: PathLike,
+        local_path: Optional[PathLike] = None,
+        *,
+        overwrite: bool = False,
+        encoding: Optional[str] = None,
+        _skip_dir_check: bool = False,
+    ) -> Optional[Union[bytes, str]]:
+        """
+        Internal method to download the content of a file path.
+
+        Parameters
+        ----------
+        path : Path or str
+            Path to the file
+        local_path : Path or str
+            Path to local file target location
+        overwrite : bool, optional
+            Should an existing file be overwritten if it exists?
+        encoding : str, optional
+            Encoding used to convert the resulting data
+        _skip_dir_check : bool, optional
+            Skip the directory check (internal use only)
+
+        Returns
+        -------
+        bytes or str - ``local_path`` is None
+        None - ``local_path`` is a Path or str
+
         """
         if local_path is not None and not overwrite and os.path.exists(local_path):
             raise OSError('target file already exists; use overwrite=True to replace')
@@ -1070,7 +1108,7 @@ def download_folder(
             remote_path = os.path.join(path, rel_path)
             target_file = os.path.normpath(os.path.join(local_path, rel_path))
             os.makedirs(os.path.dirname(target_file), exist_ok=True)
-            self.download_file(remote_path, target_file, overwrite=overwrite, _skip_dir_check=True)
+            self._download_file(remote_path, target_file, overwrite=overwrite, _skip_dir_check=True)
 
     def remove(self, path: PathLike) -> None:
         """

From 18a1f4c27cb26aac572bd7c7efd13110f28de696 Mon Sep 17 00:00:00 2001
From: Aditya Jha <ajha@singlestore.com>
Date: Thu, 18 Dec 2025 17:50:52 +0530
Subject: [PATCH 4/9] fix ci

---
 singlestoredb/management/files.py | 80 +++++++++++++++++++++++++------
 1 file changed, 65 insertions(+), 15 deletions(-)

diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py
index e62851fde..790ac4570 100644
--- a/singlestoredb/management/files.py
+++ b/singlestoredb/management/files.py
@@ -10,9 +10,12 @@
 from abc import ABC
 from abc import abstractmethod
 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
@@ -427,7 +430,7 @@ def listdir(
         path: PathLike = '/',
         *,
         recursive: bool = False,
-    ) -> List[str]:
+    ) -> Union[List[str], List[Union[str, Dict[str, Any]]]]:
         pass
 
     @abstractmethod
@@ -908,7 +911,11 @@ def is_file(self, path: PathLike) -> bool:
                 return False
             raise
 
-    def _listdir(self, path: PathLike, *, recursive: bool = False, return_meta: bool = False) -> List[Union[str, Dict[str, Any]]]:
+    def _listdir(
+        self, path: PathLike, *,
+        recursive: bool = False,
+        return_meta: bool = False,
+    ) -> List[Union[str, Dict[str, Any]]]:
         """
         Return the names (or metadata) of files in a directory.
 
@@ -929,24 +936,55 @@ def _listdir(self, path: PathLike, *, recursive: bool = False, return_meta: bool
             out: List[Union[str, Dict[str, Any]]] = []
             for item in res.get('content') or []:
                 if return_meta:
-                    out.append({'path': item['path'], 'type': item['type']})
+                    out.append(
+                        {'path': item['path'], 'type': item['type']},
+                    )
                 else:
                     out.append(item['path'])
                 if item['type'] == 'directory':
-                    out.extend(self._listdir(item['path'], recursive=recursive, return_meta=return_meta))
+                    out.extend(
+                        self._listdir(
+                            item['path'],
+                            recursive=recursive,
+                            return_meta=return_meta,
+                        ),
+                    )
             return out
 
         if return_meta:
-            return [{'path': x['path'], 'type': x['type']} for x in (res.get('content') or [])]
+            return [
+                {'path': x['path'], 'type': x['type']}
+                for x in (res.get('content') or [])
+            ]
         return [x['path'] for x in (res.get('content') or [])]
 
+    @overload
+    def listdir(
+        self,
+        path: PathLike = '/',
+        *,
+        recursive: bool = False,
+        return_meta: Literal[True] = ...,
+    ) -> List[Dict[str, Any]]:
+        ...
+
+    @overload
+    def listdir(
+        self,
+        path: PathLike = '/',
+        *,
+        recursive: bool = False,
+        return_meta: Literal[False] = ...,
+    ) -> List[str]:
+        ...
+
     def listdir(
         self,
         path: PathLike = '/',
         *,
         recursive: bool = False,
         return_meta: bool = False,
-    ) -> List[Union[str, Dict[str, Any]]]:
+    ) -> Union[List[str], List[Dict[str, Any]]]:
         """
         List the files / folders at the given path.
 
@@ -972,16 +1010,23 @@ def listdir(
         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:
-                for i in range(len(out)):
-                    if isinstance(out[i], dict):
-                        rel = '/'.join(out[i]['path'].split('/')[path_n:])
-                        out[i]['path'] = rel
-            else:
-                out = ['/'.join(str(x).split('/')[path_n:]) for x in out]
-        return out
+                result: List[Dict[str, Any]] = []
+                for item in out:
+                    if isinstance(item, dict):
+                        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)
+        return cast(List[str], out)
 
     def download_file(
         self,
@@ -1106,9 +1151,14 @@ def download_folder(
                 os.makedirs(target_dir, exist_ok=True)
                 continue
             remote_path = os.path.join(path, rel_path)
-            target_file = os.path.normpath(os.path.join(local_path, rel_path))
+            target_file = os.path.normpath(
+                os.path.join(local_path, rel_path),
+            )
             os.makedirs(os.path.dirname(target_file), exist_ok=True)
-            self._download_file(remote_path, target_file, overwrite=overwrite, _skip_dir_check=True)
+            self._download_file(
+                remote_path, target_file,
+                overwrite=overwrite, _skip_dir_check=True,
+            )
 
     def remove(self, path: PathLike) -> None:
         """

From a2af567983c8830c20d4fde6393fb6bcf9f5cd6f Mon Sep 17 00:00:00 2001
From: Aditya Jha <ajha@singlestore.com>
Date: Thu, 18 Dec 2025 17:55:45 +0530
Subject: [PATCH 5/9] fix ci mypy

---
 singlestoredb/fusion/handlers/files.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/singlestoredb/fusion/handlers/files.py b/singlestoredb/fusion/handlers/files.py
index d823aeaec..b657b8a42 100644
--- a/singlestoredb/fusion/handlers/files.py
+++ b/singlestoredb/fusion/handlers/files.py
@@ -32,6 +32,7 @@ def run(self, params: Dict[str, Any]) -> Optional[FusionSQLResult]:
             for x in file_space.listdir(
                 params['at_path'] or '/',
                 recursive=params['recursive'],
+                return_meta=False,
             ):
                 info = file_space.info(x)
                 files.append(
@@ -47,6 +48,7 @@ def run(self, params: Dict[str, Any]) -> Optional[FusionSQLResult]:
             res.set_rows([(x,) for x in file_space.listdir(
                 params['at_path'] or '/',
                 recursive=params['recursive'],
+                return_meta=False,
             )])
 
         if params['like']:

From b37b2878abd3b258555efd5d06d8c11b5f4c7fcb Mon Sep 17 00:00:00 2001
From: Aditya Jha <ajha@singlestore.com>
Date: Mon, 5 Jan 2026 19:04:02 +0530
Subject: [PATCH 6/9] review comments

---
 singlestoredb/management/files.py     | 93 ++++++++++++++++-----------
 singlestoredb/management/workspace.py | 82 ++++++++++++++++++++---
 2 files changed, 129 insertions(+), 46 deletions(-)

diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py
index 790ac4570..f66bf3a9d 100644
--- a/singlestoredb/management/files.py
+++ b/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,10 +937,10 @@ 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
         ----------
@@ -925,20 +948,18 @@ def _listdir(
             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':
@@ -946,14 +967,14 @@ def _listdir(
                         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,12 +1014,12 @@ 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))
@@ -1006,26 +1027,26 @@ def listdir(
 
         # 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)
diff --git a/singlestoredb/management/workspace.py b/singlestoredb/management/workspace.py
index 6df98a122..fd543dce7 100644
--- a/singlestoredb/management/workspace.py
+++ b/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,9 +450,13 @@ 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
         ----------
@@ -457,26 +464,64 @@ def _listdir(self, stage_path: PathLike, *, recursive: bool = False) -> List[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.
 
@@ -484,21 +529,38 @@ def listdir(
         ----------
         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))

From 12e40daf61c548ff678cf291a974ac723278c809 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Mon, 5 Jan 2026 13:53:43 +0000
Subject: [PATCH 7/9] Initial plan


From 8a5a79cd10f991aa94aa6faece92115e148ad327 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Mon, 5 Jan 2026 13:59:10 +0000
Subject: [PATCH 8/9] Address PR review comments: fix parameter names,
 decorators, and exception handling

Co-authored-by: ajha-ss <214951400+ajha-ss@users.noreply.github.com>
---
 singlestoredb/fusion/handlers/files.py | 4 ++--
 singlestoredb/management/files.py      | 4 +---
 singlestoredb/management/workspace.py  | 2 ++
 3 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/singlestoredb/fusion/handlers/files.py b/singlestoredb/fusion/handlers/files.py
index b657b8a42..7f848611b 100644
--- a/singlestoredb/fusion/handlers/files.py
+++ b/singlestoredb/fusion/handlers/files.py
@@ -32,7 +32,7 @@ def run(self, params: Dict[str, Any]) -> Optional[FusionSQLResult]:
             for x in file_space.listdir(
                 params['at_path'] or '/',
                 recursive=params['recursive'],
-                return_meta=False,
+                return_objects=False,
             ):
                 info = file_space.info(x)
                 files.append(
@@ -48,7 +48,7 @@ def run(self, params: Dict[str, Any]) -> Optional[FusionSQLResult]:
             res.set_rows([(x,) for x in file_space.listdir(
                 params['at_path'] or '/',
                 recursive=params['recursive'],
-                return_meta=False,
+                return_objects=False,
             )])
 
         if params['like']:
diff --git a/singlestoredb/management/files.py b/singlestoredb/management/files.py
index f66bf3a9d..593f7e398 100644
--- a/singlestoredb/management/files.py
+++ b/singlestoredb/management/files.py
@@ -425,7 +425,6 @@ def is_file(self, path: PathLike) -> bool:
         pass
 
     @overload
-    @abstractmethod
     def listdir(
         self,
         path: PathLike = '/',
@@ -436,7 +435,6 @@ def listdir(
         pass
 
     @overload
-    @abstractmethod
     def listdir(
         self,
         path: PathLike = '/',
@@ -1028,7 +1026,7 @@ def listdir(
         # Validate via listing GET; if response lacks 'content', it's not a directory
         try:
             out = self._listdir(path, recursive=recursive, return_objects=return_objects)
-        except Exception as exc:
+        except (ManagementError, KeyError) 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
 
diff --git a/singlestoredb/management/workspace.py b/singlestoredb/management/workspace.py
index fd543dce7..1b5d7c278 100644
--- a/singlestoredb/management/workspace.py
+++ b/singlestoredb/management/workspace.py
@@ -529,6 +529,8 @@ def listdir(
         ----------
         stage_path : Path or str, optional
             Path to the stage location
+        recursive : bool, optional
+            If True, recursively list all files and folders
         return_objects : bool, optional
             If True, return list of FilesObject instances. Otherwise just paths.
 

From 2579c2505ba9632d297748c67734f4994ffe3d8c Mon Sep 17 00:00:00 2001
From: Aditya Jha <ajha@singlestore.com>
Date: Tue, 6 Jan 2026 19:34:26 +0530
Subject: [PATCH 9/9] add tests

---
 singlestoredb/tests/test_management.py | 54 ++++++++++++++++++++++++++
 1 file changed, 54 insertions(+)

diff --git a/singlestoredb/tests/test_management.py b/singlestoredb/tests/test_management.py
index dd176f3de..f450e1f13 100755
--- a/singlestoredb/tests/test_management.py
+++ b/singlestoredb/tests/test_management.py
@@ -757,6 +757,60 @@ def test_os_directories(self):
         with self.assertRaises(s2.ManagementError):
             st.removedirs(mkdir_test_sql)
 
+    def test_listdir_return_objects(self):
+        st = self.wg.stage
+
+        listdir_test_dir = f'listdir_test_{id(self)}'
+        listdir_test_sql = f'listdir_test_{id(self)}.sql'
+
+        # Create test directory structure
+        st.mkdir(listdir_test_dir)
+        st.mkdir(f'{listdir_test_dir}/nest_1')
+        st.upload_file(TEST_DIR / 'test.sql', listdir_test_sql)
+        st.upload_file(
+            TEST_DIR / 'test.sql',
+            f'{listdir_test_dir}/nested_test.sql',
+        )
+
+        # Test return_objects=False (default behavior)
+        out = st.listdir('/')
+        assert isinstance(out, list)
+        assert all(isinstance(item, str) for item in out)
+        assert f'{listdir_test_dir}/' in out
+        assert listdir_test_sql in out
+
+        # Test return_objects=True
+        out_objs = st.listdir('/', return_objects=True)
+        assert isinstance(out_objs, list)
+        assert all(hasattr(item, 'path') for item in out_objs)
+        assert all(hasattr(item, 'type') for item in out_objs)
+
+        # Verify we have the expected items
+        obj_paths = [obj.path for obj in out_objs]
+        assert f'{listdir_test_dir}/' in obj_paths
+        assert listdir_test_sql in obj_paths
+
+        # Verify object types
+        for obj in out_objs:
+            if obj.path == f'{listdir_test_dir}/':
+                assert obj.type == 'directory'
+            elif obj.path == listdir_test_sql:
+                assert obj.type == 'file'
+
+        # Test with subdirectory and return_objects=True
+        out_objs_sub = st.listdir(listdir_test_dir, return_objects=True)
+        assert isinstance(out_objs_sub, list)
+        obj_paths_sub = [obj.path for obj in out_objs_sub]
+        assert 'nest_1/' in obj_paths_sub
+        assert 'nested_test.sql' in obj_paths_sub
+
+        # Test recursive with return_objects=True
+        out_objs_rec = st.listdir('/', recursive=True, return_objects=True)
+        obj_paths_rec = [obj.path for obj in out_objs_rec]
+        assert f'{listdir_test_dir}/' in obj_paths_rec
+        assert f'{listdir_test_dir}/nest_1/' in obj_paths_rec
+        assert f'{listdir_test_dir}/nested_test.sql' in obj_paths_rec
+
     def test_os_files(self):
         st = self.wg.stage