Ryan Kegel
eb775ba7d8
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 3m0s
- Add backend routines management with add, get, update, delete, and list functionalities. - Create models for Routine, RoutineItem, RoutineSchedule, and RoutineExtension. - Develop event types for routine confirmation and modification. - Implement frontend components for routine assignment, confirmation dialog, and routine management views. - Add unit tests for routine API and integration tests for routine CRUD flow. - Create end-to-end test plan for routines feature covering parent and child interactions.
127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
from flask import Blueprint, request, jsonify
|
|
from tinydb import Query
|
|
|
|
from api.utils import get_validated_user_id, send_event_for_current_user
|
|
from db.db import routine_db
|
|
from db.routine_items import add_item, delete_item, get_item, get_items_for_routine, update_item
|
|
from events.types.event import Event
|
|
from events.types.event_types import EventType
|
|
from events.types.routine_modified import RoutineModified
|
|
from models.routine_item import RoutineItem
|
|
|
|
routine_item_api = Blueprint('routine_item_api', __name__)
|
|
|
|
|
|
def _validate_routine_owned_by_user(routine_id: str, user_id: str):
|
|
q = Query()
|
|
return routine_db.get((q.id == routine_id) & ((q.user_id == user_id) | (q.user_id == None)))
|
|
|
|
|
|
@routine_item_api.route('/routine/<routine_id>/item/add', methods=['PUT'])
|
|
def add_routine_item(routine_id):
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
|
|
routine = _validate_routine_owned_by_user(routine_id, user_id)
|
|
if not routine:
|
|
return jsonify({'error': 'Routine not found'}), 404
|
|
|
|
data = request.get_json() or {}
|
|
name = data.get('name', '').strip()
|
|
image_id = data.get('image_id')
|
|
|
|
if not name:
|
|
return jsonify({'error': 'name is required'}), 400
|
|
|
|
existing_items = get_items_for_routine(routine_id)
|
|
order = data.get('order', len(existing_items))
|
|
|
|
item = RoutineItem(routine_id=routine_id, name=name, image_id=image_id, order=order)
|
|
add_item(item)
|
|
|
|
send_event_for_current_user(
|
|
Event(EventType.ROUTINE_MODIFIED.value, RoutineModified(routine_id, RoutineModified.OPERATION_EDIT))
|
|
)
|
|
return jsonify(item.to_dict()), 201
|
|
|
|
|
|
@routine_item_api.route('/routine/<routine_id>/item/<item_id>/edit', methods=['PUT'])
|
|
def edit_routine_item(routine_id, item_id):
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
|
|
routine = _validate_routine_owned_by_user(routine_id, user_id)
|
|
if not routine:
|
|
return jsonify({'error': 'Routine not found'}), 404
|
|
|
|
existing = get_item(item_id)
|
|
if not existing or existing.routine_id != routine_id:
|
|
return jsonify({'error': 'Item not found'}), 404
|
|
|
|
data = request.get_json(force=True) or {}
|
|
is_dirty = False
|
|
|
|
if 'name' in data:
|
|
name = data.get('name', '').strip()
|
|
if not name:
|
|
return jsonify({'error': 'name cannot be empty'}), 400
|
|
existing.name = name
|
|
is_dirty = True
|
|
|
|
if 'image_id' in data:
|
|
existing.image_id = data.get('image_id')
|
|
is_dirty = True
|
|
|
|
if 'order' in data:
|
|
order = data.get('order')
|
|
if not isinstance(order, int) or order < 0:
|
|
return jsonify({'error': 'order must be a non-negative integer'}), 400
|
|
existing.order = order
|
|
is_dirty = True
|
|
|
|
if not is_dirty:
|
|
return jsonify({'error': 'No valid fields to update'}), 400
|
|
|
|
update_item(existing)
|
|
send_event_for_current_user(
|
|
Event(EventType.ROUTINE_MODIFIED.value, RoutineModified(routine_id, RoutineModified.OPERATION_EDIT))
|
|
)
|
|
return jsonify(existing.to_dict()), 200
|
|
|
|
|
|
@routine_item_api.route('/routine/<routine_id>/item/<item_id>', methods=['DELETE'])
|
|
def delete_routine_item(routine_id, item_id):
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
|
|
routine = _validate_routine_owned_by_user(routine_id, user_id)
|
|
if not routine:
|
|
return jsonify({'error': 'Routine not found'}), 404
|
|
|
|
item = get_item(item_id)
|
|
if not item or item.routine_id != routine_id:
|
|
return jsonify({'error': 'Item not found'}), 404
|
|
|
|
delete_item(item_id)
|
|
send_event_for_current_user(
|
|
Event(EventType.ROUTINE_MODIFIED.value, RoutineModified(routine_id, RoutineModified.OPERATION_EDIT))
|
|
)
|
|
return jsonify({'message': 'Item deleted'}), 200
|
|
|
|
|
|
@routine_item_api.route('/routine/<routine_id>/items', methods=['GET'])
|
|
def list_routine_items(routine_id):
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
|
|
routine = _validate_routine_owned_by_user(routine_id, user_id)
|
|
if not routine:
|
|
return jsonify({'error': 'Routine not found'}), 404
|
|
|
|
items = [item.to_dict() for item in get_items_for_routine(routine_id)]
|
|
return jsonify({'items': items, 'count': len(items)}), 200
|