from typing import List, Optional from sqlalchemy.orm import Session from sqlalchemy import func from app.models.tag import Tag from app.models.todo import Todo from app.schemas.tag import TagCreate, TagUpdate def get_tag(db: Session, tag_id: int) -> Optional[Tag]: """Get a single tag by ID.""" return db.query(Tag).filter(Tag.id == tag_id).first() def get_tag_by_name(db: Session, name: str) -> Optional[Tag]: """Get a tag by name.""" return db.query(Tag).filter(Tag.name == name).first() def get_tags(db: Session, skip: int = 0, limit: int = 100) -> List[Tag]: """Get multiple tags with pagination.""" return db.query(Tag).offset(skip).limit(limit).all() def get_tags_count(db: Session) -> int: """Get total count of tags.""" return db.query(func.count(Tag.id)).scalar() def create_tag(db: Session, tag: TagCreate) -> Tag: """Create a new tag.""" db_tag = Tag(name=tag.name, color=tag.color) db.add(db_tag) db.commit() db.refresh(db_tag) return db_tag def update_tag(db: Session, tag_id: int, tag: TagUpdate) -> Optional[Tag]: """Update an existing tag.""" db_tag = db.query(Tag).filter(Tag.id == tag_id).first() if db_tag: update_data = tag.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(db_tag, field, value) db.commit() db.refresh(db_tag) return db_tag def delete_tag(db: Session, tag_id: int) -> bool: """Delete a tag.""" db_tag = db.query(Tag).filter(Tag.id == tag_id).first() if db_tag: db.delete(db_tag) db.commit() return True return False def add_tag_to_todo(db: Session, todo_id: int, tag_id: int) -> bool: """Add a tag to a todo.""" todo = db.query(Todo).filter(Todo.id == todo_id).first() tag = db.query(Tag).filter(Tag.id == tag_id).first() if todo and tag and tag not in todo.tags: todo.tags.append(tag) db.commit() return True return False def remove_tag_from_todo(db: Session, todo_id: int, tag_id: int) -> bool: """Remove a tag from a todo.""" todo = db.query(Todo).filter(Todo.id == todo_id).first() tag = db.query(Tag).filter(Tag.id == tag_id).first() if todo and tag and tag in todo.tags: todo.tags.remove(tag) db.commit() return True return False def get_todos_by_tag( db: Session, tag_id: int, skip: int = 0, limit: int = 100 ) -> List[Todo]: """Get all todos that have a specific tag.""" tag = db.query(Tag).filter(Tag.id == tag_id).first() if tag: return tag.todos[skip : skip + limit] return []