d993db2f17
- Add parent_id field to Todo model with self-referential foreign key
- Add parent/children relationships and is_subtask property
- Update TodoCreate/TodoUpdate schemas to include parent_id
- Add subtasks list to Todo schema and create SubtaskCreate schema
- Enhance get_todos CRUD function with parent_id filtering
- Add subtask-specific CRUD functions: get_subtasks, create_subtask, move_subtask
- Add API endpoints for subtask management
- Create migration for adding parent_id column
- Update imports and fix circular dependencies
- Ensure proper cycle prevention and validation
Features added:
- GET /todos/{todo_id}/subtasks - Get all subtasks for a todo
- POST /todos/{todo_id}/subtasks - Create a new subtask
- PUT /subtasks/{subtask_id}/move - Move subtask or convert to main todo
- Query parameter parent_id for filtering by parent
- Query parameter include_subtasks for excluding subtasks from main list
80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
from typing import List, Optional, Tuple
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.project import Project, ProjectStatus
|
|
from app.models.todo import Todo
|
|
from app.schemas.project import ProjectCreate, ProjectUpdate
|
|
|
|
|
|
def get_project(db: Session, project_id: int) -> Optional[Project]:
|
|
return db.query(Project).filter(Project.id == project_id).first()
|
|
|
|
|
|
def get_projects(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
status: Optional[ProjectStatus] = None,
|
|
search: Optional[str] = None,
|
|
) -> Tuple[List[Project], int]:
|
|
query = db.query(Project)
|
|
|
|
# Apply filters
|
|
if status is not None:
|
|
query = query.filter(Project.status == status)
|
|
if search:
|
|
query = query.filter(
|
|
Project.name.contains(search) | Project.description.contains(search)
|
|
)
|
|
|
|
# Get total count before pagination
|
|
total = query.count()
|
|
|
|
# Apply pagination and ordering
|
|
projects = query.order_by(Project.created_at.desc()).offset(skip).limit(limit).all()
|
|
|
|
return projects, total
|
|
|
|
|
|
def create_project(db: Session, project: ProjectCreate) -> Project:
|
|
db_project = Project(**project.model_dump())
|
|
db.add(db_project)
|
|
db.commit()
|
|
db.refresh(db_project)
|
|
return db_project
|
|
|
|
|
|
def update_project(
|
|
db: Session, project_id: int, project_update: ProjectUpdate
|
|
) -> Optional[Project]:
|
|
db_project = db.query(Project).filter(Project.id == project_id).first()
|
|
if db_project:
|
|
update_data = project_update.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_project, field, value)
|
|
db.commit()
|
|
db.refresh(db_project)
|
|
return db_project
|
|
|
|
|
|
def delete_project(db: Session, project_id: int) -> bool:
|
|
db_project = db.query(Project).filter(Project.id == project_id).first()
|
|
if db_project:
|
|
db.delete(db_project)
|
|
db.commit()
|
|
return True
|
|
return False
|
|
|
|
|
|
def get_project_todos(db: Session, project_id: int) -> List[Todo]:
|
|
return db.query(Todo).filter(Todo.project_id == project_id).all()
|
|
|
|
|
|
def archive_project(db: Session, project_id: int) -> Optional[Project]:
|
|
db_project = db.query(Project).filter(Project.id == project_id).first()
|
|
if db_project:
|
|
db_project.status = ProjectStatus.ARCHIVED
|
|
db.commit()
|
|
db.refresh(db_project)
|
|
return db_project
|