kanban-app/backend/app/routes/kanban.py

726 lines
No EOL
21 KiB
Python

from datetime import UTC, datetime
from flask import Blueprint, jsonify, request
from flask_jwt_extended import jwt_required, get_jwt_identity
from app import db
from app.models import (
Board,
Card,
CardLabel,
CheckItem,
Checklist,
Comment,
Label,
List,
User,
)
kanban_bp = Blueprint("kanban", __name__)
# =============================================================================
# BOARD ROUTES
# =============================================================================
@kanban_bp.route("/boards", methods=["GET"])
@jwt_required()
def get_boards():
"""Get all boards for current user"""
user_id = int(get_jwt_identity())
boards = Board.query.filter_by(user_id=user_id).all()
return jsonify([board.to_dict() for board in boards]), 200
@kanban_bp.route("/boards/<int:board_id>", methods=["GET"])
@jwt_required()
def get_board(board_id):
"""Get a single board with all its details"""
user_id = int(get_jwt_identity())
board = db.session.get(Board, board_id)
if not board:
return jsonify({"error": "Board not found"}), 404
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
# Get all lists for this board
lists_data = []
for lst in board.lists.filter_by(closed=False).order_by(List.pos).all():
cards_data = []
for card in lst.cards.filter_by(closed=False).order_by(Card.pos).all():
card_dict = card.to_dict()
# Add labels for this card
card_dict["labels"] = [
label.to_dict()
for label in (
db.session.query(Label)
.join(CardLabel)
.filter(CardLabel.card_id == card.id)
.all()
)
]
cards_data.append(card_dict)
lists_data.append({**lst.to_dict(), "cards": cards_data})
board_dict = board.to_dict()
board_dict["lists"] = lists_data
board_dict["labels"] = [label.to_dict() for label in board.labels.all()]
return jsonify(board_dict), 200
@kanban_bp.route("/boards", methods=["POST"])
@jwt_required()
def create_board():
"""Create a new board"""
user_id = int(get_jwt_identity())
data = request.get_json()
if not data or not data.get("name"):
return jsonify({"error": "Board name is required"}), 400
board = Board(
name=data["name"],
description=data.get("description"),
user_id=user_id,
url=data.get("url"),
short_link=data.get("short_link"),
short_url=data.get("short_url"),
prefs=data.get("prefs"),
label_names=data.get("label_names"),
limits=data.get("limits"),
)
db.session.add(board)
db.session.commit()
return jsonify(board.to_dict()), 201
@kanban_bp.route("/boards/<int:board_id>", methods=["PUT"])
@jwt_required()
def update_board(board_id):
"""Update a board"""
user_id = int(get_jwt_identity())
board = db.session.get(Board, board_id)
if not board:
return jsonify({"error": "Board not found"}), 404
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
board.name = data.get("name", board.name)
board.description = data.get("description", board.description)
board.closed = data.get("closed", board.closed)
board.url = data.get("url", board.url)
board.short_url = data.get("short_url", board.short_url)
board.prefs = data.get("prefs", board.prefs)
board.label_names = data.get("label_names", board.label_names)
board.limits = data.get("limits", board.limits)
board.date_last_activity = datetime.now(UTC)
db.session.commit()
return jsonify(board.to_dict()), 200
@kanban_bp.route("/boards/<int:board_id>", methods=["DELETE"])
@jwt_required()
def delete_board(board_id):
"""Delete a board"""
user_id = int(get_jwt_identity())
board = db.session.get(Board, board_id)
if not board:
return jsonify({"error": "Board not found"}), 404
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
db.session.delete(board)
db.session.commit()
return jsonify({"message": "Board deleted"}), 200
# =============================================================================
# LIST ROUTES
# =============================================================================
@kanban_bp.route("/boards/<int:board_id>/lists", methods=["POST"])
@jwt_required()
def create_list(board_id):
"""Create a new list in a board"""
user_id = int(get_jwt_identity())
board = db.session.get(Board, board_id)
if not board:
return jsonify({"error": "Board not found"}), 404
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
if not data or not data.get("name"):
return jsonify({"error": "List name is required"}), 400
lst = List(name=data["name"], board_id=board_id, pos=data.get("pos", 65535.0))
db.session.add(lst)
db.session.commit()
return jsonify(lst.to_dict()), 201
@kanban_bp.route("/lists/<int:list_id>", methods=["PUT"])
@jwt_required()
def update_list(list_id):
"""Update a list"""
user_id = int(get_jwt_identity())
lst = db.session.get(List, list_id)
if not lst:
return jsonify({"error": "List not found"}), 404
board = db.session.get(Board, lst.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
lst.name = data.get("name", lst.name)
lst.closed = data.get("closed", lst.closed)
lst.pos = data.get("pos", lst.pos)
db.session.commit()
return jsonify(lst.to_dict()), 200
@kanban_bp.route("/lists/<int:list_id>", methods=["DELETE"])
@jwt_required()
def delete_list(list_id):
"""Delete a list"""
user_id = int(get_jwt_identity())
lst = db.session.get(List, list_id)
if not lst:
return jsonify({"error": "List not found"}), 404
board = db.session.get(Board, lst.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
db.session.delete(lst)
db.session.commit()
return jsonify({"message": "List deleted"}), 200
# =============================================================================
# CARD ROUTES
# =============================================================================
@kanban_bp.route("/lists/<int:list_id>/cards", methods=["POST"])
@jwt_required()
def create_card(list_id):
"""Create a new card in a list"""
user_id = int(get_jwt_identity())
lst = db.session.get(List, list_id)
if not lst:
return jsonify({"error": "List not found"}), 404
board = db.session.get(Board, lst.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
if not data or not data.get("name"):
return jsonify({"error": "Card name is required"}), 400
card = Card(
name=data["name"],
description=data.get("description"),
board_id=lst.board_id,
list_id=list_id,
pos=data.get("pos", 65535.0),
due=data.get("due"),
due_complete=data.get("due_complete", False),
badges=data.get("badges"),
cover=data.get("cover"),
desc_data=data.get("desc_data"),
)
# Update board activity
board.date_last_activity = datetime.now(UTC)
db.session.add(card)
db.session.commit()
return jsonify(card.to_dict()), 201
@kanban_bp.route("/cards/<int:card_id>", methods=["GET"])
@jwt_required()
def get_card(card_id):
"""Get a single card with full details"""
user_id = int(get_jwt_identity())
card = db.session.get(Card, card_id)
if not card:
return jsonify({"error": "Card not found"}), 404
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
card_dict = card.to_dict()
# Add labels
card_dict["labels"] = [
label.to_dict()
for label in (
db.session.query(Label)
.join(CardLabel)
.filter(CardLabel.card_id == card.id)
.all()
)
]
# Add checklists
card_dict["checklists"] = [
{**checklist.to_dict(), "items": [item.to_dict() for item in checklist.check_items.all()]}
for checklist in card.checklists.all()
]
# Add comments
card_dict["comments"] = []
for comment in card.comments.all():
comment_dict = comment.to_dict()
user = db.session.get(User, comment.user_id)
comment_dict["user"] = user.to_dict() if user else None
card_dict["comments"].append(comment_dict)
return jsonify(card_dict), 200
@kanban_bp.route("/cards/<int:card_id>", methods=["PUT"])
@jwt_required()
def update_card(card_id):
"""Update a card"""
user_id = int(get_jwt_identity())
card = db.session.get(Card, card_id)
if not card:
return jsonify({"error": "Card not found"}), 404
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
card.name = data.get("name", card.name)
card.description = data.get("description", card.description)
card.closed = data.get("closed", card.closed)
card.pos = data.get("pos", card.pos)
card.due = data.get("due")
card.due_complete = data.get("due_complete", card.due_complete)
card.badges = data.get("badges", card.badges)
card.cover = data.get("cover", card.cover)
card.desc_data = data.get("desc_data", card.desc_data)
# Handle moving card to different list
if "list_id" in data:
new_list_id = data["list_id"]
new_list = db.session.get(List, new_list_id)
if new_list and new_list.board_id == card.board_id:
card.list_id = new_list_id
card.date_last_activity = datetime.now(UTC)
board.date_last_activity = datetime.now(UTC)
db.session.commit()
return jsonify(card.to_dict()), 200
@kanban_bp.route("/cards/<int:card_id>", methods=["DELETE"])
@jwt_required()
def delete_card(card_id):
"""Delete a card"""
user_id = int(get_jwt_identity())
card = db.session.get(Card, card_id)
if not card:
return jsonify({"error": "Card not found"}), 404
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
db.session.delete(card)
db.session.commit()
return jsonify({"message": "Card deleted"}), 200
# =============================================================================
# LABEL ROUTES
# =============================================================================
@kanban_bp.route("/boards/<int:board_id>/labels", methods=["GET"])
@jwt_required()
def get_labels(board_id):
"""Get all labels for a board"""
user_id = int(get_jwt_identity())
board = db.session.get(Board, board_id)
if not board:
return jsonify({"error": "Board not found"}), 404
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
labels = board.labels.all()
return jsonify([label.to_dict() for label in labels]), 200
@kanban_bp.route("/boards/<int:board_id>/labels", methods=["POST"])
@jwt_required()
def create_label(board_id):
"""Create a new label in a board"""
user_id = int(get_jwt_identity())
board = db.session.get(Board, board_id)
if not board:
return jsonify({"error": "Board not found"}), 404
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
if not data or not data.get("name"):
return jsonify({"error": "Label name is required"}), 400
label = Label(
name=data["name"],
color=data.get("color"),
board_id=board_id,
)
db.session.add(label)
db.session.commit()
return jsonify(label.to_dict()), 201
@kanban_bp.route("/cards/<int:card_id>/labels", methods=["POST"])
@jwt_required()
def add_label_to_card(card_id):
"""Add a label to a card"""
user_id = int(get_jwt_identity())
card = db.session.get(Card, card_id)
if not card:
return jsonify({"error": "Card not found"}), 404
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
label_id = data.get("label_id")
if not label_id:
return jsonify({"error": "Label ID is required"}), 400
label = db.session.get(Label, label_id)
if not label or label.board_id != card.board_id:
return jsonify({"error": "Label not found or doesn't belong to this board"}), 404
# Check if already labeled
existing = CardLabel.query.filter_by(card_id=card_id, label_id=label_id).first()
if existing:
return jsonify({"error": "Label already added to card"}), 400
card_label = CardLabel(card_id=card_id, label_id=label_id)
db.session.add(card_label)
db.session.commit()
return jsonify(card_label.to_dict()), 201
@kanban_bp.route("/cards/<int:card_id>/labels/<int:label_id>", methods=["DELETE"])
@jwt_required()
def remove_label_from_card(card_id, label_id):
"""Remove a label from a card"""
user_id = int(get_jwt_identity())
card = db.session.get(Card, card_id)
if not card:
return jsonify({"error": "Card not found"}), 404
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
card_label = CardLabel.query.filter_by(card_id=card_id, label_id=label_id).first()
if not card_label:
return jsonify({"error": "Label not found on card"}), 404
db.session.delete(card_label)
db.session.commit()
return jsonify({"message": "Label removed from card"}), 200
# =============================================================================
# CHECKLIST ROUTES
# =============================================================================
@kanban_bp.route("/cards/<int:card_id>/checklists", methods=["POST"])
@jwt_required()
def create_checklist(card_id):
"""Create a new checklist in a card"""
user_id = int(get_jwt_identity())
card = db.session.get(Card, card_id)
if not card:
return jsonify({"error": "Card not found"}), 404
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
if not data or not data.get("name"):
return jsonify({"error": "Checklist name is required"}), 400
checklist = Checklist(
name=data["name"],
card_id=card_id,
board_id=card.board_id,
pos=data.get("pos", 0),
)
db.session.add(checklist)
db.session.commit()
return jsonify(checklist.to_dict()), 201
@kanban_bp.route("/checklists/<int:checklist_id>/items", methods=["POST"])
@jwt_required()
def create_check_item(checklist_id):
"""Create a new check item in a checklist"""
user_id = int(get_jwt_identity())
checklist = db.session.get(Checklist, checklist_id)
if not checklist:
return jsonify({"error": "Checklist not found"}), 404
card = db.session.get(Card, checklist.card_id)
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
if not data or not data.get("name"):
return jsonify({"error": "Check item name is required"}), 400
check_item = CheckItem(
name=data["name"],
checklist_id=checklist_id,
pos=data.get("pos", 0),
state=data.get("state", "incomplete"),
due=data.get("due"),
)
db.session.add(check_item)
db.session.commit()
return jsonify(check_item.to_dict()), 201
@kanban_bp.route("/check-items/<int:item_id>", methods=["PUT"])
@jwt_required()
def update_check_item(item_id):
"""Update a check item"""
user_id = int(get_jwt_identity())
check_item = db.session.get(CheckItem, item_id)
if not check_item:
return jsonify({"error": "Check item not found"}), 404
checklist = db.session.get(Checklist, check_item.checklist_id)
card = db.session.get(Card, checklist.card_id)
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
check_item.name = data.get("name", check_item.name)
check_item.state = data.get("state", check_item.state)
check_item.pos = data.get("pos", check_item.pos)
check_item.due = data.get("due")
db.session.commit()
return jsonify(check_item.to_dict()), 200
@kanban_bp.route("/checklists/<int:checklist_id>", methods=["DELETE"])
@jwt_required()
def delete_checklist(checklist_id):
"""Delete a checklist"""
user_id = int(get_jwt_identity())
checklist = db.session.get(Checklist, checklist_id)
if not checklist:
return jsonify({"error": "Checklist not found"}), 404
card = db.session.get(Card, checklist.card_id)
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
db.session.delete(checklist)
db.session.commit()
return jsonify({"message": "Checklist deleted"}), 200
@kanban_bp.route("/check-items/<int:item_id>", methods=["DELETE"])
@jwt_required()
def delete_check_item(item_id):
"""Delete a check item"""
user_id = int(get_jwt_identity())
check_item = db.session.get(CheckItem, item_id)
if not check_item:
return jsonify({"error": "Check item not found"}), 404
checklist = db.session.get(Checklist, check_item.checklist_id)
card = db.session.get(Card, checklist.card_id)
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
db.session.delete(check_item)
db.session.commit()
return jsonify({"message": "Check item deleted"}), 200
# =============================================================================
# COMMENT ROUTES
# =============================================================================
@kanban_bp.route("/cards/<int:card_id>/comments", methods=["GET"])
@jwt_required()
def get_comments(card_id):
"""Get all comments for a card"""
user_id = int(get_jwt_identity())
card = db.session.get(Card, card_id)
if not card:
return jsonify({"error": "Card not found"}), 404
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
comments = []
for comment in card.comments.order_by(Comment.created_at.desc()).all():
comment_dict = comment.to_dict()
user = db.session.get(User, comment.user_id)
comment_dict["user"] = user.to_dict() if user else None
comments.append(comment_dict)
return jsonify(comments), 200
@kanban_bp.route("/cards/<int:card_id>/comments", methods=["POST"])
@jwt_required()
def create_comment(card_id):
"""Create a new comment on a card"""
user_id = int(get_jwt_identity())
card = db.session.get(Card, card_id)
if not card:
return jsonify({"error": "Card not found"}), 404
board = db.session.get(Board, card.board_id)
if board.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
if not data or not data.get("text"):
return jsonify({"error": "Comment text is required"}), 400
comment = Comment(text=data["text"], card_id=card_id, user_id=user_id)
db.session.add(comment)
db.session.commit()
return jsonify(comment.to_dict()), 201
@kanban_bp.route("/comments/<int:comment_id>", methods=["PUT"])
@jwt_required()
def update_comment(comment_id):
"""Update a comment"""
user_id = int(get_jwt_identity())
comment = db.session.get(Comment, comment_id)
if not comment:
return jsonify({"error": "Comment not found"}), 404
if comment.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
data = request.get_json()
comment.text = data.get("text", comment.text)
db.session.commit()
return jsonify(comment.to_dict()), 200
@kanban_bp.route("/comments/<int:comment_id>", methods=["DELETE"])
@jwt_required()
def delete_comment(comment_id):
"""Delete a comment"""
user_id = int(get_jwt_identity())
comment = db.session.get(Comment, comment_id)
if not comment:
return jsonify({"error": "Comment not found"}), 404
if comment.user_id != user_id:
return jsonify({"error": "Access denied"}), 403
db.session.delete(comment)
db.session.commit()
return jsonify({"message": "Comment deleted"}), 200