103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
"""This module handles CRUD operations for todo-items in the database, based on pydanctic schemas."""
|
|
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from todo.models import todos as todomodel
|
|
from todo.models import users as usermodel
|
|
from todo.schemas import todos as todoschema
|
|
from todo.utils.exceptions import NotFoundException, InvalidFilterParameterException
|
|
|
|
|
|
def create_todo(db: Session, todo: todoschema.TodoItemCreate, user_id: int) -> todoschema.TodoItem:
|
|
"""Creates the specified todo-item for the user with the specified id in the database."""
|
|
|
|
db_user = db.query(usermodel.User).filter(usermodel.User.id == user_id).first()
|
|
if not db_user:
|
|
raise NotFoundException(f"User with id '{user_id}' not found.")
|
|
|
|
db_todo = todomodel.TodoItem(
|
|
user_id=user_id,
|
|
title=todo.title,
|
|
description=todo.description
|
|
)
|
|
|
|
db.add(db_todo)
|
|
db.commit()
|
|
db.refresh(db_todo)
|
|
return todoschema.TodoItem.from_orm(db_todo)
|
|
|
|
|
|
def read_todo(db: Session, todo_id: int) -> todoschema.TodoItem:
|
|
"""Queries the db for a todo-item with the specified id and returns it."""
|
|
|
|
db_todo = db.query(todomodel.TodoItem).filter(todomodel.TodoItem.id == todo_id).first()
|
|
if not db_todo:
|
|
raise NotFoundException(f"Todo item with id '{todo_id}' not found.")
|
|
|
|
return todoschema.TodoItem.from_orm(db_todo)
|
|
|
|
|
|
def read_todos_for_user(db: Session, user_id: int, skip: int = 0, limit: int = 100) -> list[todoschema.TodoItem]:
|
|
"""Returns a range of todo-items of the user with the specified user_id from the database."""
|
|
|
|
for parameter in [skip, limit]:
|
|
if not isinstance(parameter, int):
|
|
raise InvalidFilterParameterException(f"Parameter '{parameter}' must be an integer.")
|
|
if parameter < 0:
|
|
raise InvalidFilterParameterException(f"Parameter '{parameter}' cannot be smaller than zero.")
|
|
|
|
db_user = db.query(usermodel.User).filter(usermodel.User.id == user_id).first()
|
|
if not db_user:
|
|
raise NotFoundException(f"User with id '{user_id}' not found.")
|
|
|
|
db_todos = db.query(todomodel.TodoItem).filter(todomodel.TodoItem.user_id == user_id).offset(skip).limit(limit).all()
|
|
return [todoschema.TodoItem.from_orm(db_todo) for db_todo in db_todos]
|
|
|
|
|
|
def read_todos_count_for_user(db: Session, user_id: int) -> int:
|
|
"""Returns the total number of todo-items of the user with the specified user_id."""
|
|
|
|
db_user = db.query(usermodel.User).filter(usermodel.User.id == user_id).first()
|
|
if not db_user:
|
|
raise NotFoundException(f"User with id '{user_id}' not found.")
|
|
|
|
return db.query(todomodel.TodoItem).filter(todomodel.TodoItem.user_id == user_id).count()
|
|
|
|
|
|
def update_todo(db: Session, todo: todoschema.TodoItemUpdate, todo_id: int) -> todoschema.TodoItem:
|
|
"""Updates the todo-item with the provided id with all non-None fields from the input todo-item."""
|
|
|
|
db_todo = db.query(todomodel.TodoItem).filter(todomodel.TodoItem.id == todo_id).first()
|
|
if not db_todo:
|
|
raise NotFoundException(f"Todo item with id '{todo_id}' not found.")
|
|
db_todo_orm_obj = todoschema.TodoItem.from_orm(db_todo)
|
|
|
|
for key in ['title', 'description', 'done']:
|
|
value = getattr(todo, key)
|
|
if value is not None:
|
|
setattr(db_todo, key, value)
|
|
# If we just newly finished a TODO, record the time it was finished
|
|
if todo.done and not db_todo_orm_obj.done:
|
|
setattr(db_todo, "finished", datetime.now(db_todo_orm_obj.updated.tzinfo))
|
|
|
|
db.commit()
|
|
db.refresh(db_todo)
|
|
return todoschema.TodoItem.from_orm(db_todo)
|
|
|
|
|
|
def delete_todo(db: Session, todo_id: int) -> todoschema.TodoItem:
|
|
"""Deletes the todo-item with the provided id from the db."""
|
|
|
|
db_todo = db.query(todomodel.TodoItem).filter(todomodel.TodoItem.id == todo_id).first()
|
|
if not db_todo:
|
|
raise NotFoundException(f"Todo item with id '{todo_id}' not found.")
|
|
todo_copy = todoschema.TodoItem.from_orm(db_todo)
|
|
|
|
db.delete(db_todo)
|
|
db.commit()
|
|
|
|
todo_copy.updated = datetime.now(todo_copy.updated.tzinfo)
|
|
return todo_copy
|