133 lines
3.9 KiB
Python
133 lines
3.9 KiB
Python
"""
|
|
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()
|
|
)
|