278 lines
8.1 KiB
Python
278 lines
8.1 KiB
Python
"""Board sharing API endpoints."""
|
|
|
|
from datetime import datetime
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.boards.schemas import (
|
|
BoardDetail,
|
|
CommentCreate,
|
|
CommentResponse,
|
|
ShareLinkCreate,
|
|
ShareLinkResponse,
|
|
)
|
|
from app.boards.sharing import generate_secure_token
|
|
from app.core.deps import get_current_user, get_db
|
|
from app.database.models.board import Board
|
|
from app.database.models.comment import Comment
|
|
from app.database.models.share_link import ShareLink
|
|
from app.database.models.user import User
|
|
|
|
router = APIRouter(tags=["sharing"])
|
|
|
|
|
|
def validate_share_link(token: str, db: Session, required_permission: str = "view-only") -> ShareLink:
|
|
"""
|
|
Validate share link token and check permissions.
|
|
|
|
Args:
|
|
token: Share link token
|
|
db: Database session
|
|
required_permission: Required permission level
|
|
|
|
Returns:
|
|
ShareLink if valid
|
|
|
|
Raises:
|
|
HTTPException: 403 if invalid or insufficient permissions
|
|
"""
|
|
share_link = (
|
|
db.query(ShareLink)
|
|
.filter(
|
|
ShareLink.token == token,
|
|
ShareLink.is_revoked == False, # noqa: E712
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if share_link is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Invalid or revoked share link",
|
|
)
|
|
|
|
# Check expiration
|
|
if share_link.expires_at and share_link.expires_at < datetime.utcnow():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Share link has expired",
|
|
)
|
|
|
|
# Check permission level
|
|
if required_permission == "view-comment" and share_link.permission_level != "view-comment":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Insufficient permissions - commenting not allowed",
|
|
)
|
|
|
|
# Update access tracking
|
|
share_link.access_count += 1
|
|
share_link.last_accessed_at = datetime.utcnow()
|
|
db.commit()
|
|
|
|
return share_link
|
|
|
|
|
|
@router.post("/boards/{board_id}/share-links", response_model=ShareLinkResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_share_link(
|
|
board_id: UUID,
|
|
share_link_data: ShareLinkCreate,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> ShareLinkResponse:
|
|
"""
|
|
Create a new share link for a board.
|
|
|
|
Only the board owner can create share links.
|
|
"""
|
|
# Verify board exists and user owns it
|
|
board = db.query(Board).filter(Board.id == board_id, Board.user_id == current_user.id).first()
|
|
|
|
if board is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found or access denied",
|
|
)
|
|
|
|
# Generate unique token
|
|
token = generate_secure_token()
|
|
|
|
# Create share link
|
|
share_link = ShareLink(
|
|
board_id=board_id,
|
|
token=token,
|
|
permission_level=share_link_data.permission_level,
|
|
expires_at=share_link_data.expires_at,
|
|
)
|
|
db.add(share_link)
|
|
db.commit()
|
|
db.refresh(share_link)
|
|
|
|
return ShareLinkResponse.model_validate(share_link)
|
|
|
|
|
|
@router.get("/boards/{board_id}/share-links", response_model=list[ShareLinkResponse])
|
|
def list_share_links(
|
|
board_id: UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> list[ShareLinkResponse]:
|
|
"""
|
|
List all share links for a board.
|
|
|
|
Only the board owner can list share links.
|
|
"""
|
|
# Verify board exists and user owns it
|
|
board = db.query(Board).filter(Board.id == board_id, Board.user_id == current_user.id).first()
|
|
|
|
if board is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found or access denied",
|
|
)
|
|
|
|
# Get all share links for board
|
|
share_links = db.query(ShareLink).filter(ShareLink.board_id == board_id).order_by(ShareLink.created_at.desc()).all()
|
|
|
|
return [ShareLinkResponse.model_validate(link) for link in share_links]
|
|
|
|
|
|
@router.delete("/boards/{board_id}/share-links/{link_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def revoke_share_link(
|
|
board_id: UUID,
|
|
link_id: UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> None:
|
|
"""
|
|
Revoke (soft delete) a share link.
|
|
|
|
Only the board owner can revoke share links.
|
|
"""
|
|
# Verify board exists and user owns it
|
|
board = db.query(Board).filter(Board.id == board_id, Board.user_id == current_user.id).first()
|
|
|
|
if board is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found or access denied",
|
|
)
|
|
|
|
# Get and revoke share link
|
|
share_link = db.query(ShareLink).filter(ShareLink.id == link_id, ShareLink.board_id == board_id).first()
|
|
|
|
if share_link is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Share link not found",
|
|
)
|
|
|
|
share_link.is_revoked = True
|
|
db.commit()
|
|
|
|
|
|
@router.get("/shared/{token}", response_model=BoardDetail)
|
|
def get_shared_board(
|
|
token: str,
|
|
db: Session = Depends(get_db),
|
|
) -> BoardDetail:
|
|
"""
|
|
Access a shared board via token.
|
|
|
|
No authentication required - access controlled by share link token.
|
|
"""
|
|
# Validate share link
|
|
share_link = validate_share_link(token, db, required_permission="view-only")
|
|
|
|
# Get board details
|
|
board = db.query(Board).filter(Board.id == share_link.board_id).first()
|
|
|
|
if board is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found",
|
|
)
|
|
|
|
return BoardDetail.model_validate(board)
|
|
|
|
|
|
@router.post("/shared/{token}/comments", response_model=CommentResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_comment(
|
|
token: str,
|
|
comment_data: CommentCreate,
|
|
db: Session = Depends(get_db),
|
|
) -> CommentResponse:
|
|
"""
|
|
Create a comment on a shared board.
|
|
|
|
Requires view-comment permission level.
|
|
"""
|
|
# Validate share link with comment permission
|
|
share_link = validate_share_link(token, db, required_permission="view-comment")
|
|
|
|
# Create comment
|
|
comment = Comment(
|
|
board_id=share_link.board_id,
|
|
share_link_id=share_link.id,
|
|
author_name=comment_data.author_name,
|
|
content=comment_data.content,
|
|
position=comment_data.position,
|
|
)
|
|
db.add(comment)
|
|
db.commit()
|
|
db.refresh(comment)
|
|
|
|
return CommentResponse.model_validate(comment)
|
|
|
|
|
|
@router.get("/shared/{token}/comments", response_model=list[CommentResponse])
|
|
def list_comments(
|
|
token: str,
|
|
db: Session = Depends(get_db),
|
|
) -> list[CommentResponse]:
|
|
"""
|
|
List all comments on a shared board.
|
|
|
|
Requires view-only or view-comment permission.
|
|
"""
|
|
# Validate share link
|
|
share_link = validate_share_link(token, db, required_permission="view-only")
|
|
|
|
# Get all comments for board (non-deleted)
|
|
comments = (
|
|
db.query(Comment)
|
|
.filter(Comment.board_id == share_link.board_id, Comment.is_deleted == False) # noqa: E712
|
|
.order_by(Comment.created_at.desc())
|
|
.all()
|
|
)
|
|
|
|
return [CommentResponse.model_validate(comment) for comment in comments]
|
|
|
|
|
|
@router.get("/boards/{board_id}/comments", response_model=list[CommentResponse])
|
|
def list_board_comments(
|
|
board_id: UUID,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> list[CommentResponse]:
|
|
"""
|
|
List all comments on a board (owner view).
|
|
|
|
Only the board owner can access this endpoint.
|
|
"""
|
|
# Verify board exists and user owns it
|
|
board = db.query(Board).filter(Board.id == board_id, Board.user_id == current_user.id).first()
|
|
|
|
if board is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found or access denied",
|
|
)
|
|
|
|
# Get all comments for board (including deleted for owner)
|
|
comments = db.query(Comment).filter(Comment.board_id == board_id).order_by(Comment.created_at.desc()).all()
|
|
|
|
return [CommentResponse.model_validate(comment) for comment in comments]
|