101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from tinydb import Query
|
|
|
|
from events.sse import send_event_to_user
|
|
from events.types.event import Event
|
|
from events.types.event_types import EventType
|
|
from events.types.task_modified import TaskModified
|
|
from models.task import Task
|
|
from db.db import task_db, child_db
|
|
|
|
task_api = Blueprint('task_api', __name__)
|
|
|
|
# Task endpoints
|
|
@task_api.route('/task/add', methods=['PUT'])
|
|
def add_task():
|
|
data = request.get_json()
|
|
name = data.get('name')
|
|
points = data.get('points')
|
|
is_good = data.get('is_good')
|
|
image = data.get('image_id', '')
|
|
if not name or points is None or is_good is None:
|
|
return jsonify({'error': 'Name, points, and is_good are required'}), 400
|
|
task = Task(name=name, points=points, is_good=is_good, image_id=image)
|
|
task_db.insert(task.to_dict())
|
|
send_event_to_user("user123", Event(EventType.TASK_MODIFIED.value,
|
|
TaskModified(task.id, TaskModified.OPERATION_ADD)))
|
|
return jsonify({'message': f'Task {name} added.'}), 201
|
|
|
|
@task_api.route('/task/<id>', methods=['GET'])
|
|
def get_task(id):
|
|
TaskQuery = Query()
|
|
result = task_db.search(TaskQuery.id == id)
|
|
if not result:
|
|
return jsonify({'error': 'Task not found'}), 404
|
|
return jsonify(result[0]), 200
|
|
|
|
@task_api.route('/task/list', methods=['GET'])
|
|
def list_tasks():
|
|
tasks = task_db.all()
|
|
return jsonify({'tasks': tasks}), 200
|
|
|
|
@task_api.route('/task/<id>', methods=['DELETE'])
|
|
def delete_task(id):
|
|
TaskQuery = Query()
|
|
removed = task_db.remove(TaskQuery.id == id)
|
|
if removed:
|
|
# remove the task id from any child's task list
|
|
ChildQuery = Query()
|
|
for child in child_db.all():
|
|
tasks = child.get('tasks', [])
|
|
if id in tasks:
|
|
tasks.remove(id)
|
|
child_db.update({'tasks': tasks}, ChildQuery.id == child.get('id'))
|
|
send_event_to_user("user123", Event(EventType.TASK_MODIFIED.value,
|
|
TaskModified(id, TaskModified.OPERATION_DELETE)))
|
|
|
|
return jsonify({'message': f'Task {id} deleted.'}), 200
|
|
return jsonify({'error': 'Task not found'}), 404
|
|
|
|
@task_api.route('/task/<id>/edit', methods=['PUT'])
|
|
def edit_task(id):
|
|
TaskQuery = Query()
|
|
existing = task_db.get(TaskQuery.id == id)
|
|
if not existing:
|
|
return jsonify({'error': 'Task not found'}), 404
|
|
|
|
data = request.get_json(force=True) or {}
|
|
updates = {}
|
|
|
|
if 'name' in data:
|
|
name = data.get('name', '').strip()
|
|
if not name:
|
|
return jsonify({'error': 'Name cannot be empty'}), 400
|
|
updates['name'] = name
|
|
|
|
if 'points' in data:
|
|
points = data.get('points')
|
|
if not isinstance(points, int):
|
|
return jsonify({'error': 'Points must be an integer'}), 400
|
|
if points <= 0:
|
|
return jsonify({'error': 'Points must be a positive integer'}), 400
|
|
updates['points'] = points
|
|
|
|
if 'is_good' in data:
|
|
is_good = data.get('is_good')
|
|
if not isinstance(is_good, bool):
|
|
return jsonify({'error': 'is_good must be a boolean'}), 400
|
|
updates['is_good'] = is_good
|
|
|
|
if 'image_id' in data:
|
|
updates['image_id'] = data.get('image_id', '')
|
|
|
|
if not updates:
|
|
return jsonify({'error': 'No valid fields to update'}), 400
|
|
|
|
task_db.update(updates, TaskQuery.id == id)
|
|
updated = task_db.get(TaskQuery.id == id)
|
|
send_event_to_user("user123", Event(EventType.TASK_MODIFIED.value,
|
|
TaskModified(id, TaskModified.OPERATION_EDIT)))
|
|
return jsonify(updated), 200
|