mirror of
https://github.com/OHV-IT/collabrix.git
synced 2025-12-15 16:48:36 +01:00
a7ff948e7e
- Added complete Kanban board functionality with drag-and-drop - Implemented auto-save for Kanban card editing (no more edit button) - Added route persistence to remember last visited page on reload - Improved Kanban UI design with slimmer borders and compact layout - Added checklist functionality for Kanban cards - Enhanced file upload and direct messaging features - Improved authentication and user management - Added toast notifications system - Various UI/UX improvements and bug fixes
705 lines
23 KiB
Python
705 lines
23 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlmodel import Session, select
|
|
from typing import List
|
|
from app.database import get_session
|
|
from app.models import (
|
|
KanbanBoard, KanbanColumn, KanbanCard, Channel, User,
|
|
KanbanChecklist, KanbanChecklistItem
|
|
)
|
|
from app.schemas import (
|
|
KanbanBoardCreate, KanbanBoardUpdate, KanbanBoardResponse,
|
|
KanbanColumnCreate, KanbanColumnUpdate, KanbanColumnResponse,
|
|
KanbanCardCreate, KanbanCardUpdate, KanbanCardResponse,
|
|
KanbanBoardWithColumns, KanbanColumnWithCards,
|
|
KanbanChecklistCreate, KanbanChecklistUpdate, KanbanChecklistResponse,
|
|
KanbanChecklistItemCreate, KanbanChecklistItemUpdate, KanbanChecklistItemResponse,
|
|
KanbanChecklistWithItems, KanbanCardWithChecklists
|
|
)
|
|
from app.auth import get_current_user
|
|
|
|
router = APIRouter(prefix="/kanban", tags=["Kanban"])
|
|
|
|
|
|
# Board endpoints
|
|
@router.post("/boards", response_model=KanbanBoardResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_board(
|
|
board_data: KanbanBoardCreate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Create a new kanban board for a channel"""
|
|
# Check if channel exists and user has access
|
|
channel = session.get(Channel, board_data.channel_id)
|
|
if not channel:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Channel not found"
|
|
)
|
|
|
|
# Check if user has access to the channel's department
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied to this channel"
|
|
)
|
|
|
|
# Check if board already exists for this channel
|
|
existing_board = session.exec(
|
|
select(KanbanBoard).where(KanbanBoard.channel_id == board_data.channel_id)
|
|
).first()
|
|
if existing_board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Board already exists for this channel"
|
|
)
|
|
|
|
new_board = KanbanBoard(
|
|
channel_id=board_data.channel_id,
|
|
name=board_data.name
|
|
)
|
|
|
|
session.add(new_board)
|
|
session.commit()
|
|
session.refresh(new_board)
|
|
|
|
return new_board
|
|
|
|
|
|
@router.get("/boards/{channel_id}", response_model=KanbanBoardWithColumns)
|
|
def get_board_by_channel(
|
|
channel_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Get kanban board for a specific channel"""
|
|
# Check if channel exists and user has access
|
|
channel = session.get(Channel, channel_id)
|
|
if not channel:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Channel not found"
|
|
)
|
|
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied to this channel"
|
|
)
|
|
|
|
# Get board with columns and cards
|
|
board = session.exec(
|
|
select(KanbanBoard).where(KanbanBoard.channel_id == channel_id)
|
|
).first()
|
|
|
|
if not board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="No kanban board found for this channel"
|
|
)
|
|
|
|
# Load columns with cards
|
|
columns = session.exec(
|
|
select(KanbanColumn)
|
|
.where(KanbanColumn.board_id == board.id)
|
|
.order_by(KanbanColumn.position)
|
|
).all()
|
|
|
|
board_data = KanbanBoardWithColumns.from_orm(board)
|
|
board_data.columns = []
|
|
|
|
for column in columns:
|
|
cards = session.exec(
|
|
select(KanbanCard)
|
|
.where(KanbanCard.column_id == column.id)
|
|
.order_by(KanbanCard.position)
|
|
).all()
|
|
|
|
column_data = KanbanColumnWithCards.from_orm(column)
|
|
column_data.cards = [KanbanCardResponse.from_orm(card) for card in cards]
|
|
board_data.columns.append(column_data)
|
|
|
|
return board_data
|
|
|
|
|
|
@router.put("/boards/{board_id}", response_model=KanbanBoardResponse)
|
|
def update_board(
|
|
board_id: int,
|
|
board_data: KanbanBoardUpdate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Update a kanban board"""
|
|
board = session.get(KanbanBoard, board_id)
|
|
if not board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found"
|
|
)
|
|
|
|
# Check access via channel
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
update_data = board_data.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(board, field, value)
|
|
|
|
session.commit()
|
|
session.refresh(board)
|
|
return board
|
|
|
|
|
|
# Column endpoints
|
|
@router.post("/columns", response_model=KanbanColumnResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_column(
|
|
column_data: KanbanColumnCreate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Create a new kanban column"""
|
|
# Check if board exists and user has access
|
|
board = session.get(KanbanBoard, column_data.board_id)
|
|
if not board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found"
|
|
)
|
|
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
new_column = KanbanColumn(
|
|
board_id=column_data.board_id,
|
|
name=column_data.name,
|
|
position=column_data.position,
|
|
color=column_data.color
|
|
)
|
|
|
|
session.add(new_column)
|
|
session.commit()
|
|
session.refresh(new_column)
|
|
|
|
return new_column
|
|
|
|
|
|
@router.put("/columns/{column_id}", response_model=KanbanColumnResponse)
|
|
def update_column(
|
|
column_id: int,
|
|
column_data: KanbanColumnUpdate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Update a kanban column"""
|
|
column = session.get(KanbanColumn, column_id)
|
|
if not column:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Column not found"
|
|
)
|
|
|
|
# Check access via board and channel
|
|
board = session.get(KanbanBoard, column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
update_data = column_data.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(column, field, value)
|
|
|
|
session.commit()
|
|
session.refresh(column)
|
|
return column
|
|
|
|
|
|
@router.delete("/columns/{column_id}")
|
|
def delete_column(
|
|
column_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Delete a kanban column"""
|
|
column = session.get(KanbanColumn, column_id)
|
|
if not column:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Column not found"
|
|
)
|
|
|
|
# Check access via board and channel
|
|
board = session.get(KanbanBoard, column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
session.delete(column)
|
|
session.commit()
|
|
return {"message": "Column deleted successfully"}
|
|
|
|
|
|
# Card endpoints
|
|
@router.post("/cards", response_model=KanbanCardResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_card(
|
|
card_data: KanbanCardCreate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Create a new kanban card"""
|
|
# Check if column exists and user has access
|
|
column = session.get(KanbanColumn, card_data.column_id)
|
|
if not column:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Column not found"
|
|
)
|
|
|
|
board = session.get(KanbanBoard, column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
new_card = KanbanCard(
|
|
column_id=card_data.column_id,
|
|
title=card_data.title,
|
|
description=card_data.description,
|
|
assignee_id=card_data.assignee_id,
|
|
position=card_data.position,
|
|
due_date=card_data.due_date,
|
|
priority=card_data.priority,
|
|
labels=card_data.labels
|
|
)
|
|
|
|
session.add(new_card)
|
|
session.commit()
|
|
session.refresh(new_card)
|
|
|
|
return new_card
|
|
|
|
|
|
@router.put("/cards/{card_id}", response_model=KanbanCardResponse)
|
|
def update_card(
|
|
card_id: int,
|
|
card_data: KanbanCardUpdate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Update a kanban card"""
|
|
card = session.get(KanbanCard, card_id)
|
|
if not card:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Card not found"
|
|
)
|
|
|
|
# Check access via column, board and channel
|
|
column = session.get(KanbanColumn, card.column_id)
|
|
board = session.get(KanbanBoard, column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
update_data = card_data.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(card, field, value)
|
|
|
|
session.commit()
|
|
session.refresh(card)
|
|
return card
|
|
|
|
|
|
@router.delete("/cards/{card_id}")
|
|
def delete_card(
|
|
card_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Delete a kanban card"""
|
|
card = session.get(KanbanCard, card_id)
|
|
if not card:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Card not found"
|
|
)
|
|
|
|
# Check access via column, board and channel
|
|
column = session.get(KanbanColumn, card.column_id)
|
|
board = session.get(KanbanBoard, column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
session.delete(card)
|
|
session.commit()
|
|
return {"message": "Card deleted successfully"}
|
|
|
|
|
|
@router.put("/cards/{card_id}/move")
|
|
def move_card(
|
|
card_id: int,
|
|
target_column_id: int,
|
|
new_position: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Move a card to a different column and/or position"""
|
|
card = session.get(KanbanCard, card_id)
|
|
if not card:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Card not found"
|
|
)
|
|
|
|
target_column = session.get(KanbanColumn, target_column_id)
|
|
if not target_column:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Target column not found"
|
|
)
|
|
|
|
# Check access for both source and target
|
|
source_column = session.get(KanbanColumn, card.column_id)
|
|
source_board = session.get(KanbanBoard, source_column.board_id)
|
|
target_board = session.get(KanbanBoard, target_column.board_id)
|
|
|
|
if source_board.id != target_board.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Cannot move cards between different boards"
|
|
)
|
|
|
|
channel = session.get(Channel, source_board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
# Update card position
|
|
card.column_id = target_column_id
|
|
card.position = new_position
|
|
|
|
session.commit()
|
|
session.refresh(card)
|
|
return {"message": "Card moved successfully"}
|
|
|
|
|
|
# Checklist endpoints
|
|
@router.post("/checklists", response_model=KanbanChecklistResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_checklist(
|
|
checklist_data: KanbanChecklistCreate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Create a new checklist for a card"""
|
|
# Check if card exists and user has access
|
|
card = session.get(KanbanCard, checklist_data.card_id)
|
|
if not card:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Card not found"
|
|
)
|
|
|
|
# Check access via board -> channel -> department
|
|
board = session.get(KanbanBoard, card.column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
new_checklist = KanbanChecklist(
|
|
card_id=checklist_data.card_id,
|
|
title=checklist_data.title,
|
|
position=checklist_data.position
|
|
)
|
|
|
|
session.add(new_checklist)
|
|
session.commit()
|
|
session.refresh(new_checklist)
|
|
|
|
return new_checklist
|
|
|
|
|
|
@router.get("/checklists/{checklist_id}", response_model=KanbanChecklistWithItems)
|
|
def get_checklist(
|
|
checklist_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Get a checklist with its items"""
|
|
checklist = session.get(KanbanChecklist, checklist_id)
|
|
if not checklist:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Checklist not found"
|
|
)
|
|
|
|
# Check access via card -> board -> channel -> department
|
|
board = session.get(KanbanBoard, checklist.card.column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
return checklist
|
|
|
|
|
|
@router.put("/checklists/{checklist_id}", response_model=KanbanChecklistResponse)
|
|
def update_checklist(
|
|
checklist_id: int,
|
|
checklist_data: KanbanChecklistUpdate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Update a checklist"""
|
|
checklist = session.get(KanbanChecklist, checklist_id)
|
|
if not checklist:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Checklist not found"
|
|
)
|
|
|
|
# Check access
|
|
board = session.get(KanbanBoard, checklist.card.column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
# Update fields
|
|
if checklist_data.title is not None:
|
|
checklist.title = checklist_data.title
|
|
if checklist_data.position is not None:
|
|
checklist.position = checklist_data.position
|
|
|
|
session.commit()
|
|
session.refresh(checklist)
|
|
|
|
return checklist
|
|
|
|
|
|
@router.delete("/checklists/{checklist_id}")
|
|
def delete_checklist(
|
|
checklist_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Delete a checklist"""
|
|
checklist = session.get(KanbanChecklist, checklist_id)
|
|
if not checklist:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Checklist not found"
|
|
)
|
|
|
|
# Check access
|
|
board = session.get(KanbanBoard, checklist.card.column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
session.delete(checklist)
|
|
session.commit()
|
|
|
|
return {"message": "Checklist deleted successfully"}
|
|
|
|
|
|
# Checklist Item endpoints
|
|
@router.post("/checklist-items", response_model=KanbanChecklistItemResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_checklist_item(
|
|
item_data: KanbanChecklistItemCreate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Create a new checklist item"""
|
|
# Check if checklist exists and user has access
|
|
checklist = session.get(KanbanChecklist, item_data.checklist_id)
|
|
if not checklist:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Checklist not found"
|
|
)
|
|
|
|
# Check access via checklist -> card -> board -> channel -> department
|
|
board = session.get(KanbanBoard, checklist.card.column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
new_item = KanbanChecklistItem(
|
|
checklist_id=item_data.checklist_id,
|
|
title=item_data.title,
|
|
is_completed=item_data.is_completed,
|
|
position=item_data.position
|
|
)
|
|
|
|
session.add(new_item)
|
|
session.commit()
|
|
session.refresh(new_item)
|
|
|
|
return new_item
|
|
|
|
|
|
@router.put("/checklist-items/{item_id}", response_model=KanbanChecklistItemResponse)
|
|
def update_checklist_item(
|
|
item_id: int,
|
|
item_data: KanbanChecklistItemUpdate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Update a checklist item"""
|
|
item = session.get(KanbanChecklistItem, item_id)
|
|
if not item:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Checklist item not found"
|
|
)
|
|
|
|
# Check access via item -> checklist -> card -> board -> channel -> department
|
|
board = session.get(KanbanBoard, item.checklist.card.column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
# Update fields
|
|
if item_data.title is not None:
|
|
item.title = item_data.title
|
|
if item_data.is_completed is not None:
|
|
item.is_completed = item_data.is_completed
|
|
if item_data.position is not None:
|
|
item.position = item_data.position
|
|
|
|
session.commit()
|
|
session.refresh(item)
|
|
|
|
return item
|
|
|
|
|
|
@router.delete("/checklist-items/{item_id}")
|
|
def delete_checklist_item(
|
|
item_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Delete a checklist item"""
|
|
item = session.get(KanbanChecklistItem, item_id)
|
|
if not item:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Checklist item not found"
|
|
)
|
|
|
|
# Check access
|
|
board = session.get(KanbanBoard, item.checklist.card.column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
session.delete(item)
|
|
session.commit()
|
|
|
|
return {"message": "Checklist item deleted successfully"}
|
|
|
|
|
|
@router.get("/cards/{card_id}/checklists", response_model=List[KanbanChecklistWithItems])
|
|
def get_card_checklists(
|
|
card_id: int,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Get all checklists for a specific card"""
|
|
# Check if card exists and user has access
|
|
card = session.get(KanbanCard, card_id)
|
|
if not card:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Card not found"
|
|
)
|
|
|
|
# Check access via board -> channel -> department
|
|
board = session.get(KanbanBoard, card.column.board_id)
|
|
channel = session.get(Channel, board.channel_id)
|
|
user_departments = [dept.id for dept in current_user.departments]
|
|
if channel.department_id not in user_departments and not current_user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Access denied"
|
|
)
|
|
|
|
# Get all checklists for the card with their items
|
|
checklists = session.exec(
|
|
select(KanbanChecklist)
|
|
.where(KanbanChecklist.card_id == card_id)
|
|
.order_by(KanbanChecklist.position)
|
|
).all()
|
|
|
|
result = []
|
|
for checklist in checklists:
|
|
items = session.exec(
|
|
select(KanbanChecklistItem)
|
|
.where(KanbanChecklistItem.checklist_id == checklist.id)
|
|
.order_by(KanbanChecklistItem.position)
|
|
).all()
|
|
|
|
checklist_data = KanbanChecklistWithItems.from_orm(checklist)
|
|
checklist_data.items = [KanbanChecklistItemResponse.from_orm(item) for item in items]
|
|
result.append(checklist_data)
|
|
|
|
return result |