feat: Optimize directory operations for 10-100x performance improvement (#350)

Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude <noreply@anthropic.com>

Author: phernandez

src/basic_memory/api/routers/directory_router.py

@@ -31,6 +31,27 @@ async def get_directory_tree(
     return tree
 
 
+@router.get("/structure", response_model=DirectoryNode)
+async def get_directory_structure(
+    directory_service: DirectoryServiceDep,
+    project_id: ProjectIdDep,
+):
+    """Get folder structure for navigation (no files).
+
+    Optimized endpoint for folder tree navigation. Returns only directory nodes
+    without file metadata. For full tree with files, use /directory/tree.
+
+    Args:
+        directory_service: Service for directory operations
+        project_id: ID of the current project
+
+    Returns:
+        DirectoryNode tree containing only folders (type="directory")
+    """
+    structure = await directory_service.get_directory_structure()
+    return structure
+
+
 @router.get("/list", response_model=List[DirectoryNode])
 async def list_directory(
     directory_service: DirectoryServiceDep,

src/basic_memory/repository/entity_repository.py

@@ -176,6 +176,66 @@ async def upsert_entity(self, entity: Entity) -> Entity:
                     entity = await self._handle_permalink_conflict(entity, session)
                     return entity
 
+    async def get_distinct_directories(self) -> List[str]:
+        """Extract unique directory paths from file_path column.
+
+        Optimized method for getting directory structure without loading full entities
+        or relationships. Returns a sorted list of unique directory paths.
+
+        Returns:
+            List of unique directory paths (e.g., ["notes", "notes/meetings", "specs"])
+        """
+        # Query only file_path column, no entity objects or relationships
+        query = select(Entity.file_path).distinct()
+        query = self._add_project_filter(query)
+
+        # Execute with use_query_options=False to skip eager loading
+        result = await self.execute_query(query, use_query_options=False)
+        file_paths = [row for row in result.scalars().all()]
+
+        # Parse file paths to extract unique directories
+        directories = set()
+        for file_path in file_paths:
+            parts = [p for p in file_path.split("/") if p]
+            # Add all parent directories (exclude filename which is the last part)
+            for i in range(len(parts) - 1):
+                dir_path = "/".join(parts[: i + 1])
+                directories.add(dir_path)
+
+        return sorted(directories)
+
+    async def find_by_directory_prefix(self, directory_prefix: str) -> Sequence[Entity]:
+        """Find entities whose file_path starts with the given directory prefix.
+
+        Optimized method for listing directory contents without loading all entities.
+        Uses SQL LIKE pattern matching to filter entities by directory path.
+
+        Args:
+            directory_prefix: Directory path prefix (e.g., "docs", "docs/guides")
+                             Empty string returns all entities (root directory)
+
+        Returns:
+            Sequence of entities in the specified directory and subdirectories
+        """
+        # Build SQL LIKE pattern
+        if directory_prefix == "" or directory_prefix == "/":
+            # Root directory - return all entities
+            return await self.find_all()
+
+        # Remove leading/trailing slashes for consistency
+        directory_prefix = directory_prefix.strip("/")
+
+        # Query entities with file_path starting with prefix
+        # Pattern matches "prefix/" to ensure we get files IN the directory,
+        # not just files whose names start with the prefix
+        pattern = f"{directory_prefix}/%"
+
+        query = self.select().where(Entity.file_path.like(pattern))
+
+        # Skip eager loading - we only need basic entity fields for directory trees
+        result = await self.execute_query(query, use_query_options=False)
+        return list(result.scalars().all())
+
     async def _handle_permalink_conflict(self, entity: Entity, session: AsyncSession) -> Entity:
         """Handle permalink conflicts by generating a unique permalink."""
         base_permalink = entity.permalink

src/basic_memory/repository/repository.py

@@ -152,12 +152,25 @@ def select(self, *entities: Any) -> Select:
         # Add project filter if applicable
         return self._add_project_filter(query)
 
-    async def find_all(self, skip: int = 0, limit: Optional[int] = None) -> Sequence[T]:
-        """Fetch records from the database with pagination."""
+    async def find_all(
+        self, skip: int = 0, limit: Optional[int] = None, use_load_options: bool = True
+    ) -> Sequence[T]:
+        """Fetch records from the database with pagination.
+
+        Args:
+            skip: Number of records to skip
+            limit: Maximum number of records to return
+            use_load_options: Whether to apply eager loading options (default: True)
+        """
         logger.debug(f"Finding all {self.Model.__name__} (skip={skip}, limit={limit})")
 
         async with db.scoped_session(self.session_maker) as session:
-            query = select(self.Model).offset(skip).options(*self.get_load_options())
+            query = select(self.Model).offset(skip)
+
+            # Only apply load options if requested
+            if use_load_options:
+                query = query.options(*self.get_load_options())
+
             # Add project filter if applicable
             query = self._add_project_filter(query)
 

src/basic_memory/services/directory_service.py

@@ -3,8 +3,9 @@
 import fnmatch
 import logging
 import os
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Sequence
 
+from basic_memory.models import Entity
 from basic_memory.repository import EntityRepository
 from basic_memory.schemas.directory import DirectoryNode
 
@@ -89,6 +90,49 @@ async def get_directory_tree(self) -> DirectoryNode:
         # Return the root node with its children
         return root_node
 
+    async def get_directory_structure(self) -> DirectoryNode:
+        """Build a hierarchical directory structure without file details.
+
+        Optimized method for folder navigation that only returns directory nodes,
+        no file metadata. Much faster than get_directory_tree() for large knowledge bases.
+
+        Returns:
+            DirectoryNode tree containing only folders (type="directory")
+        """
+        # Get unique directories without loading entities
+        directories = await self.entity_repository.get_distinct_directories()
+
+        # Create a root directory node
+        root_node = DirectoryNode(name="Root", directory_path="/", type="directory")
+
+        # Map to store directory nodes by path for easy lookup
+        dir_map: Dict[str, DirectoryNode] = {"/": root_node}
+
+        # Build tree with just folders
+        for dir_path in directories:
+            parts = [p for p in dir_path.split("/") if p]
+            current_path = "/"
+
+            for i, part in enumerate(parts):
+                parent_path = current_path
+                # Build the directory path
+                current_path = (
+                    f"{current_path}{part}" if current_path == "/" else f"{current_path}/{part}"
+                )
+
+                # Create directory node if it doesn't exist
+                if current_path not in dir_map:
+                    dir_node = DirectoryNode(
+                        name=part, directory_path=current_path, type="directory"
+                    )
+                    dir_map[current_path] = dir_node
+
+                    # Add to parent's children
+                    if parent_path in dir_map:
+                        dir_map[parent_path].children.append(dir_node)
+
+        return root_node
+
     async def list_directory(
         self,
         dir_name: str = "/",
@@ -118,8 +162,13 @@ async def list_directory(
         if dir_name != "/" and dir_name.endswith("/"):
             dir_name = dir_name.rstrip("/")
 
-        # Get the full directory tree
-        root_tree = await self.get_directory_tree()
+        # Optimize: Query only entities in the target directory
+        # instead of loading the entire tree
+        dir_prefix = dir_name.lstrip("/")
+        entity_rows = await self.entity_repository.find_by_directory_prefix(dir_prefix)
+
+        # Build a partial tree from only the relevant entities
+        root_tree = self._build_directory_tree_from_entities(entity_rows, dir_name)
 
         # Find the target directory node
         target_node = self._find_directory_node(root_tree, dir_name)
@@ -132,6 +181,78 @@ async def list_directory(
 
         return result
 
+    def _build_directory_tree_from_entities(
+        self, entity_rows: Sequence[Entity], root_path: str
+    ) -> DirectoryNode:
+        """Build a directory tree from a subset of entities.
+
+        Args:
+            entity_rows: Sequence of entity objects to build tree from
+            root_path: Root directory path for the tree
+
+        Returns:
+            DirectoryNode representing the tree root
+        """
+        # Create a root directory node
+        root_node = DirectoryNode(name="Root", directory_path=root_path, type="directory")
+
+        # Map to store directory nodes by path for easy lookup
+        dir_map: Dict[str, DirectoryNode] = {root_path: root_node}
+
+        # First pass: create all directory nodes
+        for file in entity_rows:
+            # Process directory path components
+            parts = [p for p in file.file_path.split("/") if p]
+
+            # Create directory structure
+            current_path = "/"
+            for i, part in enumerate(parts[:-1]):  # Skip the filename
+                parent_path = current_path
+                # Build the directory path
+                current_path = (
+                    f"{current_path}{part}" if current_path == "/" else f"{current_path}/{part}"
+                )
+
+                # Create directory node if it doesn't exist
+                if current_path not in dir_map:
+                    dir_node = DirectoryNode(
+                        name=part, directory_path=current_path, type="directory"
+                    )
+                    dir_map[current_path] = dir_node
+
+                    # Add to parent's children
+                    if parent_path in dir_map:
+                        dir_map[parent_path].children.append(dir_node)
+
+        # Second pass: add file nodes to their parent directories
+        for file in entity_rows:
+            file_name = os.path.basename(file.file_path)
+            parent_dir = os.path.dirname(file.file_path)
+            directory_path = "/" if parent_dir == "" else f"/{parent_dir}"
+
+            # Create file node
+            file_node = DirectoryNode(
+                name=file_name,
+                file_path=file.file_path,
+                directory_path=f"/{file.file_path}",
+                type="file",
+                title=file.title,
+                permalink=file.permalink,
+                entity_id=file.id,
+                entity_type=file.entity_type,
+                content_type=file.content_type,
+                updated_at=file.updated_at,
+            )
+
+            # Add to parent directory's children
+            if directory_path in dir_map:
+                dir_map[directory_path].children.append(file_node)
+            elif root_path in dir_map:
+                # Fallback to root if parent not found
+                dir_map[root_path].children.append(file_node)
+
+        return root_node
+
     def _find_directory_node(
         self, root: DirectoryNode, target_path: str
     ) -> Optional[DirectoryNode]:

src/basic_memory/services/project_service.py

@@ -80,7 +80,13 @@ def current_project(self) -> str:
         return os.environ.get("BASIC_MEMORY_PROJECT", self.config_manager.default_project)
 
     async def list_projects(self) -> Sequence[Project]:
-        return await self.repository.find_all()
+        """List all projects without loading entity relationships.
+
+        Returns only basic project fields (name, path, etc.) without
+        eager loading the entities relationship which could load thousands
+        of entities for large knowledge bases.
+        """
+        return await self.repository.find_all(use_load_options=False)
 
     async def get_project(self, name: str) -> Optional[Project]:
         """Get the file path for a project by name or permalink."""