228 lines
6.5 KiB
Python
228 lines
6.5 KiB
Python
"""Routes for Wiki CRUD operations."""
|
|
|
|
import re
|
|
|
|
from flask_jwt_extended import get_jwt_identity, jwt_required
|
|
from flask_pydantic import validate
|
|
from sqlalchemy import and_
|
|
from sqlalchemy import delete as sql_delete
|
|
from sqlalchemy import select
|
|
|
|
from app import db
|
|
from app.models import Board, Wiki, wiki_entity_links
|
|
from app.models.card import Card
|
|
from app.models.epic import Epic
|
|
from app.schemas import (CardResponse, EpicResponse, WikiCreateRequest,
|
|
WikiLinkRequest, WikiResponse, WikiUpdateRequest)
|
|
|
|
from . import kanban_bp
|
|
|
|
|
|
def generate_slug(name):
|
|
"""Generate URL-friendly slug from name."""
|
|
slug = name.lower()
|
|
slug = re.sub(r"[^a-z0-9\s-]", "", slug)
|
|
slug = re.sub(r"\s+", "-", slug)
|
|
slug = slug.strip("-")
|
|
return slug
|
|
|
|
|
|
@kanban_bp.route("/boards/<int:board_id>/wikis", methods=["GET"])
|
|
@jwt_required()
|
|
def get_board_wikis(board_id):
|
|
"""Get all wikis for a board."""
|
|
board = db.session.get(Board, board_id)
|
|
if not board:
|
|
return {"error": "Board not found"}, 404
|
|
|
|
wikis = Wiki.query.filter_by(board_id=board_id).all()
|
|
return [WikiResponse.model_validate(wiki).model_dump() for wiki in wikis], 200
|
|
|
|
|
|
@kanban_bp.route("/boards/<int:board_id>/wikis", methods=["POST"])
|
|
@jwt_required()
|
|
@validate(body=WikiCreateRequest)
|
|
def create_wiki(board_id, body: WikiCreateRequest):
|
|
"""Create a new wiki for a board."""
|
|
board = db.session.get(Board, board_id)
|
|
if not board:
|
|
return {"error": "Board not found"}, 404
|
|
|
|
# Generate slug if not provided
|
|
slug = body.slug
|
|
if not slug:
|
|
slug = generate_slug(body.name)
|
|
|
|
# Ensure uniqueness
|
|
counter = 1
|
|
original_slug = slug
|
|
while Wiki.query.filter_by(slug=slug).first():
|
|
slug = f"{original_slug}-{counter}"
|
|
counter += 1
|
|
|
|
wiki = Wiki(
|
|
board_id=board_id,
|
|
name=body.name,
|
|
slug=slug,
|
|
content=body.content,
|
|
summary=body.summary,
|
|
category=body.category,
|
|
tags=body.tags,
|
|
created_by=get_jwt_identity(),
|
|
updated_by=get_jwt_identity(),
|
|
)
|
|
|
|
db.session.add(wiki)
|
|
db.session.commit()
|
|
|
|
return WikiResponse.model_validate(wiki).model_dump(), 201
|
|
|
|
|
|
@kanban_bp.route("/wikis/<int:wiki_id>", methods=["GET"])
|
|
@jwt_required()
|
|
def get_wiki(wiki_id):
|
|
"""Get a specific wiki with details including linked entities."""
|
|
wiki = db.session.get(Wiki, wiki_id)
|
|
if not wiki:
|
|
return {"error": "Wiki not found"}, 404
|
|
|
|
# Convert wiki to dict and add linked entities
|
|
wiki_dict = wiki.to_dict()
|
|
|
|
# Add linked cards
|
|
linked_cards = [
|
|
CardResponse.model_validate(card).model_dump() for card in wiki.linked_cards
|
|
]
|
|
|
|
# Add linked epics
|
|
linked_epics = [
|
|
EpicResponse.model_validate(epic).model_dump() for epic in wiki.linked_epics
|
|
]
|
|
|
|
wiki_dict["linked_cards"] = linked_cards
|
|
wiki_dict["linked_epics"] = linked_epics
|
|
|
|
return wiki_dict, 200
|
|
|
|
|
|
@kanban_bp.route("/wikis/<int:wiki_id>", methods=["PUT"])
|
|
@jwt_required()
|
|
@validate(body=WikiUpdateRequest)
|
|
def update_wiki(wiki_id, body: WikiUpdateRequest):
|
|
"""Update a wiki."""
|
|
wiki = db.session.get(Wiki, wiki_id)
|
|
if not wiki:
|
|
return {"error": "Wiki not found"}, 404
|
|
|
|
# Update fields
|
|
if body.name is not None:
|
|
wiki.name = body.name
|
|
if body.slug is not None:
|
|
wiki.slug = body.slug
|
|
if body.content is not None:
|
|
wiki.content = body.content
|
|
if body.summary is not None:
|
|
wiki.summary = body.summary
|
|
if body.category is not None:
|
|
wiki.category = body.category
|
|
if body.tags is not None:
|
|
wiki.tags = body.tags
|
|
|
|
wiki.updated_by = get_jwt_identity()
|
|
db.session.commit()
|
|
|
|
return WikiResponse.model_validate(wiki).model_dump(), 200
|
|
|
|
|
|
@kanban_bp.route("/wikis/<int:wiki_id>", methods=["DELETE"])
|
|
@jwt_required()
|
|
def delete_wiki(wiki_id):
|
|
"""Delete a wiki."""
|
|
wiki = db.session.get(Wiki, wiki_id)
|
|
if not wiki:
|
|
return {"error": "Wiki not found"}, 404
|
|
|
|
# Delete wiki (cascades to wiki_entity_links)
|
|
db.session.delete(wiki)
|
|
db.session.commit()
|
|
|
|
return {"message": "Wiki deleted successfully"}, 200
|
|
|
|
|
|
@kanban_bp.route("/wikis/<int:wiki_id>/links", methods=["POST"])
|
|
@jwt_required()
|
|
@validate(body=WikiLinkRequest)
|
|
def create_wiki_link(wiki_id, body: WikiLinkRequest):
|
|
"""Link a wiki to another entity (card, epic, list, or board)."""
|
|
wiki = db.session.get(Wiki, wiki_id)
|
|
if not wiki:
|
|
return {"error": "Wiki not found"}, 404
|
|
|
|
# Check if entity exists
|
|
entity = None
|
|
if body.entity_type == "card":
|
|
entity = db.session.get(Card, body.entity_id)
|
|
elif body.entity_type == "epic":
|
|
entity = db.session.get(Epic, body.entity_id)
|
|
elif body.entity_type == "list":
|
|
from app.models.list_model import List
|
|
|
|
entity = db.session.get(List, body.entity_id)
|
|
elif body.entity_type == "board":
|
|
entity = db.session.get(Board, body.entity_id)
|
|
|
|
if not entity:
|
|
return {"error": f"{body.entity_type.capitalize()} not found"}, 404
|
|
|
|
# Check if link already exists
|
|
existing_link = db.session.scalars(
|
|
select(wiki_entity_links).where(
|
|
and_(
|
|
wiki_entity_links.c.wiki_id == wiki_id,
|
|
wiki_entity_links.c.entity_type == body.entity_type,
|
|
wiki_entity_links.c.entity_id == body.entity_id,
|
|
)
|
|
)
|
|
).first()
|
|
|
|
if existing_link:
|
|
return {"error": "Link already exists"}, 400
|
|
|
|
# Create link using core insert
|
|
stmt = wiki_entity_links.insert().values(
|
|
wiki_id=wiki_id,
|
|
entity_type=body.entity_type,
|
|
entity_id=body.entity_id,
|
|
linked_by=get_jwt_identity(),
|
|
)
|
|
db.session.execute(stmt)
|
|
db.session.commit()
|
|
|
|
return {"message": "Link created successfully"}, 201
|
|
|
|
|
|
@kanban_bp.route("/wikis/<int:wiki_id>/links", methods=["DELETE"])
|
|
@jwt_required()
|
|
@validate(body=WikiLinkRequest)
|
|
def delete_wiki_link(wiki_id, body: WikiLinkRequest):
|
|
"""Delete a wiki entity link."""
|
|
wiki = db.session.get(Wiki, wiki_id)
|
|
if not wiki:
|
|
return {"error": "Wiki not found"}, 404
|
|
|
|
# Delete link using core delete
|
|
stmt = sql_delete(wiki_entity_links).where(
|
|
and_(
|
|
wiki_entity_links.c.wiki_id == wiki_id,
|
|
wiki_entity_links.c.entity_type == body.entity_type,
|
|
wiki_entity_links.c.entity_id == body.entity_id,
|
|
)
|
|
)
|
|
result = db.session.execute(stmt)
|
|
|
|
if result.rowcount == 0:
|
|
return {"error": "Link not found"}, 404
|
|
|
|
db.session.commit()
|
|
|
|
return {"message": "Link deleted successfully"}, 200
|