from sqlalchemy.orm import Session from . import models, schemas from typing import List, Optional def get_todo(db: Session, todo_id: int) -> Optional[models.Todo]: return db.query(models.Todo).filter(models.Todo.id == todo_id).first() def get_todos(db: Session, skip: int = 0, limit: int = 100) -> List[models.Todo]: return db.query(models.Todo).offset(skip).limit(limit).all() def create_todo(db: Session, todo: schemas.TodoCreate) -> models.Todo: db_todo = models.Todo(**todo.model_dump()) db.add(db_todo) db.commit() db.refresh(db_todo) return db_todo def update_todo(db: Session, todo_id: int, todo_update: schemas.TodoUpdate) -> Optional[models.Todo]: db_todo = db.query(models.Todo).filter(models.Todo.id == todo_id).first() if db_todo: update_data = todo_update.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(db_todo, field, value) db.commit() db.refresh(db_todo) return db_todo def delete_todo(db: Session, todo_id: int) -> Optional[models.Todo]: db_todo = db.query(models.Todo).filter(models.Todo.id == todo_id).first() if db_todo: db.delete(db_todo) db.commit() return db_todo