Implement database storage layer

2025-12-12 12:45:33 -06:00
parent a6df5aba5a
commit 9604540dd3
16 changed files with 1477 additions and 2 deletions
--- a/backend/app/repositories/project.py
+++ b/backend/app/repositories/project.py
@@ -0,0 +1,132 @@
+"""
+Project repository for data access operations.
+"""
+
+from typing import Optional, List, Tuple
+from sqlalchemy.orm import Session
+from sqlalchemy import func, or_, asc, desc
+from uuid import UUID
+
+from .base import BaseRepository
+from ..models import Project
+
+
+class ProjectRepository(BaseRepository[Project]):
+    """Repository for Project entity operations."""
+
+    model = Project
+
+    def get_by_name(self, name: str) -> Optional[Project]:
+        """Get project by unique name."""
+        return self.db.query(Project).filter(Project.name == name).first()
+
+    def exists_by_name(self, name: str) -> bool:
+        """Check if project with name exists."""
+        return self.db.query(
+            self.db.query(Project).filter(Project.name == name).exists()
+        ).scalar()
+
+    def list_accessible(
+        self,
+        user_id: str,
+        page: int = 1,
+        limit: int = 20,
+        search: Optional[str] = None,
+        visibility: Optional[str] = None,
+        sort: str = "name",
+        order: str = "asc",
+    ) -> Tuple[List[Project], int]:
+        """
+        List projects accessible to user with filtering and pagination.
+
+        Returns tuple of (projects, total_count).
+        """
+        # Base query - filter by access
+        query = self.db.query(Project).filter(
+            or_(Project.is_public == True, Project.created_by == user_id)
+        )
+
+        # Apply visibility filter
+        if visibility == "public":
+            query = query.filter(Project.is_public == True)
+        elif visibility == "private":
+            query = query.filter(Project.is_public == False, Project.created_by == user_id)
+
+        # Apply search filter
+        if search:
+            search_lower = search.lower()
+            query = query.filter(
+                or_(
+                    func.lower(Project.name).contains(search_lower),
+                    func.lower(Project.description).contains(search_lower)
+                )
+            )
+
+        # Get total count before pagination
+        total = query.count()
+
+        # Apply sorting
+        sort_columns = {
+            "name": Project.name,
+            "created_at": Project.created_at,
+            "updated_at": Project.updated_at,
+        }
+        sort_column = sort_columns.get(sort, Project.name)
+        if order == "desc":
+            query = query.order_by(desc(sort_column))
+        else:
+            query = query.order_by(asc(sort_column))
+
+        # Apply pagination
+        offset = (page - 1) * limit
+        projects = query.offset(offset).limit(limit).all()
+
+        return projects, total
+
+    def create_project(
+        self,
+        name: str,
+        created_by: str,
+        description: Optional[str] = None,
+        is_public: bool = True,
+    ) -> Project:
+        """Create a new project."""
+        return self.create(
+            name=name,
+            description=description,
+            is_public=is_public,
+            created_by=created_by,
+        )
+
+    def update_project(
+        self,
+        project: Project,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        is_public: Optional[bool] = None,
+    ) -> Project:
+        """Update project fields."""
+        updates = {}
+        if name is not None:
+            updates["name"] = name
+        if description is not None:
+            updates["description"] = description
+        if is_public is not None:
+            updates["is_public"] = is_public
+        return self.update(project, **updates)
+
+    def search(self, query_str: str, limit: int = 10) -> List[Project]:
+        """Search projects by name or description."""
+        search_lower = query_str.lower()
+        return (
+            self.db.query(Project)
+            .filter(
+                or_(
+                    func.lower(Project.name).contains(search_lower),
+                    func.lower(Project.description).contains(search_lower)
+                )
+            )
+            .order_by(Project.name)
+            .limit(limit)
+            .all()
+        )