diff --git a/app/api/routers/knowledge.py b/app/api/routers/knowledge.py new file mode 100644 index 0000000..783c28b --- /dev/null +++ b/app/api/routers/knowledge.py @@ -0,0 +1,735 @@ +"""Knowledge Base router with global-permission RBAC. + +Permissions (global, granted via the Role Editor; admins auto-pass): + knowledge-base.create create a knowledge base + knowledge-base.read read any knowledge base / topic / category / fact + knowledge-base.update edit a KB and its topic/category/fact structure, + and link/unlink knowledge bases to projects + knowledge-base.delete delete a knowledge base + +There is no per-KB membership model (unlike projects) — access is purely by +the four global permissions above. +""" +import re +from typing import List, Optional +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from app.core.config import get_db +from app.models import models +from app.models import knowledge as kb_models +from app.schemas import knowledge as kb_schemas +from app.api.deps import get_current_user_or_apikey + +router = APIRouter(tags=["KnowledgeBase"]) + +PERM_CREATE = "knowledge-base.create" +PERM_READ = "knowledge-base.read" +PERM_UPDATE = "knowledge-base.update" +PERM_DELETE = "knowledge-base.delete" + + +# --------------------------------------------------------------------------- +# Permission helper (global perms only) +# --------------------------------------------------------------------------- +def _require_perm(db: Session, user: models.User, perm_name: str) -> None: + if user.is_admin: + return + from app.models.role_permission import Permission, RolePermission + has = ( + db.query(Permission.id) + .join(RolePermission, RolePermission.permission_id == Permission.id) + .filter( + RolePermission.role_id == user.role_id, + Permission.name == perm_name, + ) + .first() + if user.role_id + else None + ) + if not has: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Permission denied: {perm_name} required", + ) + + +# --------------------------------------------------------------------------- +# Knowledge-base code generation (same rules as project_code) +# --------------------------------------------------------------------------- +WORD_SEGMENT_RE = re.compile(r"[A-Za-z]+") +CAMEL_RE = re.compile(r"[A-Z]+(?=[A-Z][a-z])|[A-Z]?[a-z]+|[A-Z]+") + + +def _split_words(name: str): + segments = WORD_SEGMENT_RE.findall(name or "") + words = [] + for seg in segments: + for part in CAMEL_RE.findall(seg): + if part.isupper() and len(part) > 1: + words.extend(list(part)) + else: + words.append(part) + return words + + +def _code_exists(db: Session, code: str) -> bool: + return ( + db.query(kb_models.KnowledgeBase) + .filter(kb_models.KnowledgeBase.knowledge_base_code == code) + .first() + is not None + ) + + +def _next_counter(db: Session, prefix: str, width: int) -> str: + if width <= 0: + return "" + counter = ( + db.query(kb_models.KnowledgeBaseCodeCounter) + .filter(kb_models.KnowledgeBaseCodeCounter.prefix == prefix) + .first() + ) + if not counter: + counter = kb_models.KnowledgeBaseCodeCounter(prefix=prefix, next_value=0) + db.add(counter) + db.flush() + value = counter.next_value + counter.next_value += 1 + db.flush() + return format(value, "x").upper().zfill(width) + + +def _generate_with_counter(db: Session, prefix: str, width: int) -> str: + while True: + suffix = _next_counter(db, prefix, width) + code = (prefix + suffix).upper() + if not _code_exists(db, code): + return code + + +def _generate_kb_code(db: Session, title: str) -> str: + words = _split_words(title) + if not words: + return _generate_with_counter(db, "UN", 4) + + if len(words) == 1: + letters = "".join(c for c in words[0] if c.isalpha()).upper() + if not letters: + return _generate_with_counter(db, "UN", 4) + if len(letters) >= 6: + code = letters[:6] + if _code_exists(db, code): + return _generate_with_counter(db, letters[:2], 4) + return code + prefix = letters + return _generate_with_counter(db, prefix, 6 - len(prefix)) + + total_letters = sum(len(w) for w in words) + if len(words) > 6: + code = "".join(w[0] for w in words[:6]).upper() + if _code_exists(db, code): + return _generate_with_counter(db, code[:2], 4) + return code + + if total_letters < 6: + prefix = "".join(words).upper() + return _generate_with_counter(db, prefix, 6 - len(prefix)) + + if total_letters == 6: + code = "".join(words).upper() + if _code_exists(db, code): + return _generate_with_counter(db, code[:2], 4) + return code + + # total_letters > 6: initials, then fill from a counter on collision + code = "".join(w[0] for w in words).upper()[:6] + if not _code_exists(db, code): + return code + return _generate_with_counter(db, code[:2], 4) + + +# --------------------------------------------------------------------------- +# Resolvers +# --------------------------------------------------------------------------- +def _resolve_kb(db: Session, identifier: str) -> kb_models.KnowledgeBase: + kb = None + try: + kb = db.query(kb_models.KnowledgeBase).filter(kb_models.KnowledgeBase.id == int(identifier)).first() + except (ValueError, TypeError): + kb = ( + db.query(kb_models.KnowledgeBase) + .filter(kb_models.KnowledgeBase.knowledge_base_code == str(identifier)) + .first() + ) + if not kb: + raise HTTPException(status_code=404, detail="Knowledge base not found") + return kb + + +def _resolve_project(db: Session, identifier: str) -> models.Project: + project = None + try: + project = db.query(models.Project).filter(models.Project.id == int(identifier)).first() + except (ValueError, TypeError): + project = db.query(models.Project).filter(models.Project.project_code == str(identifier)).first() + if not project: + raise HTTPException(status_code=404, detail="Project not found") + return project + + +def _get_topic(db: Session, topic_id: int) -> kb_models.KnowledgeTopic: + topic = db.query(kb_models.KnowledgeTopic).filter(kb_models.KnowledgeTopic.id == topic_id).first() + if not topic: + raise HTTPException(status_code=404, detail="Topic not found") + return topic + + +def _get_category(db: Session, category_id: int) -> kb_models.KnowledgeCategory: + cat = db.query(kb_models.KnowledgeCategory).filter(kb_models.KnowledgeCategory.id == category_id).first() + if not cat: + raise HTTPException(status_code=404, detail="Category not found") + return cat + + +def _descendant_category_ids(db: Session, category_id: int) -> List[int]: + """Return [category_id, ...all nested descendants] (deepest last).""" + collected = [category_id] + frontier = [category_id] + while frontier: + children = ( + db.query(kb_models.KnowledgeCategory.id) + .filter(kb_models.KnowledgeCategory.parent.in_(frontier)) + .all() + ) + child_ids = [c.id for c in children] + if not child_ids: + break + collected.extend(child_ids) + frontier = child_ids + return collected + + +# =========================================================================== +# Knowledge Base CRUD +# =========================================================================== +@router.post("/knowledge-bases", response_model=kb_schemas.KnowledgeBaseResponse, status_code=status.HTTP_201_CREATED) +def create_knowledge_base( + payload: kb_schemas.KnowledgeBaseCreate, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_CREATE) + kb = kb_models.KnowledgeBase( + title=payload.title, + description=payload.description, + created_by=current_user.id, + knowledge_base_code=_generate_kb_code(db, payload.title), + ) + db.add(kb) + db.commit() + db.refresh(kb) + return kb + + +@router.get("/knowledge-bases", response_model=List[kb_schemas.KnowledgeBaseResponse]) +def list_knowledge_bases( + skip: int = 0, + limit: int = 100, + project: Optional[str] = None, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + q = db.query(kb_models.KnowledgeBase) + if project is not None: + proj = _resolve_project(db, project) + linked_ids = [ + row.knowledge_base_id + for row in db.query(kb_models.ProjectKnowledgeBase.knowledge_base_id) + .filter(kb_models.ProjectKnowledgeBase.project_id == proj.id) + .all() + ] + if not linked_ids: + return [] + q = q.filter(kb_models.KnowledgeBase.id.in_(linked_ids)) + return q.order_by(kb_models.KnowledgeBase.id).offset(skip).limit(limit).all() + + +@router.get("/knowledge-bases/{kb_id}", response_model=kb_schemas.KnowledgeBaseResponse) +def get_knowledge_base( + kb_id: str, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + return _resolve_kb(db, kb_id) + + +@router.patch("/knowledge-bases/{kb_id}", response_model=kb_schemas.KnowledgeBaseResponse) +def update_knowledge_base( + kb_id: str, + payload: kb_schemas.KnowledgeBaseUpdate, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + kb = _resolve_kb(db, kb_id) + data = payload.model_dump(exclude_unset=True) + for field, value in data.items(): + setattr(kb, field, value) + db.commit() + db.refresh(kb) + return kb + + +@router.delete("/knowledge-bases/{kb_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_knowledge_base( + kb_id: str, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_DELETE) + kb = _resolve_kb(db, kb_id) + + topic_ids = [ + t.id + for t in db.query(kb_models.KnowledgeTopic.id) + .filter(kb_models.KnowledgeTopic.knowledge_base_id == kb.id) + .all() + ] + if topic_ids: + db.query(kb_models.KnowledgeFact).filter(kb_models.KnowledgeFact.topic_id.in_(topic_ids)).delete(synchronize_session=False) + db.query(kb_models.KnowledgeCategory).filter(kb_models.KnowledgeCategory.topic_id.in_(topic_ids)).delete(synchronize_session=False) + db.query(kb_models.KnowledgeTopic).filter(kb_models.KnowledgeTopic.id.in_(topic_ids)).delete(synchronize_session=False) + db.query(kb_models.ProjectKnowledgeBase).filter(kb_models.ProjectKnowledgeBase.knowledge_base_id == kb.id).delete(synchronize_session=False) + db.delete(kb) + db.commit() + return None + + +# =========================================================================== +# Tree (read-only aggregate) +# =========================================================================== +@router.get("/knowledge-bases/{kb_id}/tree", response_model=kb_schemas.KnowledgeBaseTree) +def get_knowledge_base_tree( + kb_id: str, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + kb = _resolve_kb(db, kb_id) + + topics = db.query(kb_models.KnowledgeTopic).filter(kb_models.KnowledgeTopic.knowledge_base_id == kb.id).all() + topic_ids = [t.id for t in topics] + cats = ( + db.query(kb_models.KnowledgeCategory).filter(kb_models.KnowledgeCategory.topic_id.in_(topic_ids)).all() + if topic_ids else [] + ) + facts = ( + db.query(kb_models.KnowledgeFact).filter(kb_models.KnowledgeFact.topic_id.in_(topic_ids)).all() + if topic_ids else [] + ) + + # Index facts by (topic_id, category_id) and categories by (topic_id, parent) + facts_by_cat: dict = {} + facts_topic_direct: dict = {} + for f in facts: + fr = kb_schemas.KnowledgeFactResponse.model_validate(f) + if f.category_id is None: + facts_topic_direct.setdefault(f.topic_id, []).append(fr) + else: + facts_by_cat.setdefault(f.category_id, []).append(fr) + + cats_by_parent: dict = {} + for c in cats: + cats_by_parent.setdefault((c.topic_id, c.parent), []).append(c) + + def build_category(cat) -> kb_schemas.CategoryTreeNode: + children = cats_by_parent.get((cat.topic_id, cat.id), []) + return kb_schemas.CategoryTreeNode( + id=cat.id, + name=cat.name, + parent=cat.parent, + topic_id=cat.topic_id, + description=cat.description, + categories=[build_category(ch) for ch in children], + facts=facts_by_cat.get(cat.id, []), + ) + + topic_nodes = [] + for t in topics: + top_level_cats = cats_by_parent.get((t.id, None), []) + topic_nodes.append( + kb_schemas.TopicTreeNode( + id=t.id, + topic=t.topic, + knowledge_base_id=t.knowledge_base_id, + description=t.description, + categories=[build_category(c) for c in top_level_cats], + facts=facts_topic_direct.get(t.id, []), + ) + ) + + return kb_schemas.KnowledgeBaseTree( + id=kb.id, + knowledge_base_code=kb.knowledge_base_code, + title=kb.title, + description=kb.description, + topics=topic_nodes, + ) + + +# =========================================================================== +# Topics +# =========================================================================== +@router.get("/knowledge-bases/{kb_id}/topics", response_model=List[kb_schemas.KnowledgeTopicResponse]) +def list_topics( + kb_id: str, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + kb = _resolve_kb(db, kb_id) + return db.query(kb_models.KnowledgeTopic).filter(kb_models.KnowledgeTopic.knowledge_base_id == kb.id).all() + + +@router.post("/knowledge-bases/{kb_id}/topics", response_model=kb_schemas.KnowledgeTopicResponse, status_code=status.HTTP_201_CREATED) +def create_topic( + kb_id: str, + payload: kb_schemas.KnowledgeTopicCreate, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + kb = _resolve_kb(db, kb_id) + existing = ( + db.query(kb_models.KnowledgeTopic) + .filter( + kb_models.KnowledgeTopic.knowledge_base_id == kb.id, + kb_models.KnowledgeTopic.topic == payload.topic, + ) + .first() + ) + if existing: + raise HTTPException(status_code=400, detail="A topic with this name already exists in this knowledge base") + topic = kb_models.KnowledgeTopic( + topic=payload.topic, + description=payload.description, + knowledge_base_id=kb.id, + created_by=current_user.id, + ) + db.add(topic) + db.commit() + db.refresh(topic) + return topic + + +@router.get("/knowledge-topics/{topic_id}", response_model=kb_schemas.KnowledgeTopicResponse) +def get_topic( + topic_id: int, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + return _get_topic(db, topic_id) + + +@router.patch("/knowledge-topics/{topic_id}", response_model=kb_schemas.KnowledgeTopicResponse) +def update_topic( + topic_id: int, + payload: kb_schemas.KnowledgeTopicUpdate, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + topic = _get_topic(db, topic_id) + data = payload.model_dump(exclude_unset=True) + if "topic" in data and data["topic"] and data["topic"] != topic.topic: + clash = ( + db.query(kb_models.KnowledgeTopic) + .filter( + kb_models.KnowledgeTopic.knowledge_base_id == topic.knowledge_base_id, + kb_models.KnowledgeTopic.topic == data["topic"], + kb_models.KnowledgeTopic.id != topic.id, + ) + .first() + ) + if clash: + raise HTTPException(status_code=400, detail="A topic with this name already exists in this knowledge base") + for field, value in data.items(): + setattr(topic, field, value) + db.commit() + db.refresh(topic) + return topic + + +@router.delete("/knowledge-topics/{topic_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_topic( + topic_id: int, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + topic = _get_topic(db, topic_id) + db.query(kb_models.KnowledgeFact).filter(kb_models.KnowledgeFact.topic_id == topic.id).delete(synchronize_session=False) + db.query(kb_models.KnowledgeCategory).filter(kb_models.KnowledgeCategory.topic_id == topic.id).delete(synchronize_session=False) + db.delete(topic) + db.commit() + return None + + +# =========================================================================== +# Categories +# =========================================================================== +def _check_category_unique(db: Session, topic_id: int, parent: Optional[int], name: str, exclude_id: Optional[int] = None): + q = db.query(kb_models.KnowledgeCategory).filter( + kb_models.KnowledgeCategory.topic_id == topic_id, + kb_models.KnowledgeCategory.name == name, + ) + if parent is None: + q = q.filter(kb_models.KnowledgeCategory.parent.is_(None)) + else: + q = q.filter(kb_models.KnowledgeCategory.parent == parent) + if exclude_id is not None: + q = q.filter(kb_models.KnowledgeCategory.id != exclude_id) + if q.first(): + raise HTTPException(status_code=400, detail="A category with this name already exists under the same parent") + + +@router.get("/knowledge-topics/{topic_id}/categories", response_model=List[kb_schemas.KnowledgeCategoryResponse]) +def list_categories( + topic_id: int, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + _get_topic(db, topic_id) + return db.query(kb_models.KnowledgeCategory).filter(kb_models.KnowledgeCategory.topic_id == topic_id).all() + + +@router.post("/knowledge-categories", response_model=kb_schemas.KnowledgeCategoryResponse, status_code=status.HTTP_201_CREATED) +def create_category( + payload: kb_schemas.KnowledgeCategoryCreate, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + _get_topic(db, payload.topic_id) + if payload.parent is not None: + parent_cat = _get_category(db, payload.parent) + if parent_cat.topic_id != payload.topic_id: + raise HTTPException(status_code=400, detail="Parent category belongs to a different topic") + _check_category_unique(db, payload.topic_id, payload.parent, payload.name) + cat = kb_models.KnowledgeCategory( + name=payload.name, + description=payload.description, + parent=payload.parent, + topic_id=payload.topic_id, + created_by=current_user.id, + last_updated_by=current_user.id, + ) + db.add(cat) + db.commit() + db.refresh(cat) + return cat + + +@router.get("/knowledge-categories/{category_id}", response_model=kb_schemas.KnowledgeCategoryResponse) +def get_category( + category_id: int, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + return _get_category(db, category_id) + + +@router.patch("/knowledge-categories/{category_id}", response_model=kb_schemas.KnowledgeCategoryResponse) +def update_category( + category_id: int, + payload: kb_schemas.KnowledgeCategoryUpdate, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + cat = _get_category(db, category_id) + data = payload.model_dump(exclude_unset=True) + + new_parent = data.get("parent", cat.parent) if "parent" in data else cat.parent + if "parent" in data and data["parent"] is not None: + if data["parent"] == cat.id: + raise HTTPException(status_code=400, detail="A category cannot be its own parent") + parent_cat = _get_category(db, data["parent"]) + if parent_cat.topic_id != cat.topic_id: + raise HTTPException(status_code=400, detail="Parent category belongs to a different topic") + # Prevent cycles: new parent must not be a descendant of this category + if data["parent"] in _descendant_category_ids(db, cat.id): + raise HTTPException(status_code=400, detail="Cannot move a category under one of its own descendants") + + new_name = data.get("name", cat.name) + if ("name" in data and data["name"] != cat.name) or ("parent" in data and new_parent != cat.parent): + _check_category_unique(db, cat.topic_id, new_parent, new_name, exclude_id=cat.id) + + for field, value in data.items(): + setattr(cat, field, value) + cat.last_updated_by = current_user.id + db.commit() + db.refresh(cat) + return cat + + +@router.delete("/knowledge-categories/{category_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_category( + category_id: int, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + cat = _get_category(db, category_id) + ids = _descendant_category_ids(db, cat.id) + db.query(kb_models.KnowledgeFact).filter(kb_models.KnowledgeFact.category_id.in_(ids)).delete(synchronize_session=False) + db.query(kb_models.KnowledgeCategory).filter(kb_models.KnowledgeCategory.id.in_(ids)).delete(synchronize_session=False) + db.commit() + return None + + +# =========================================================================== +# Facts +# =========================================================================== +@router.post("/knowledge-facts", response_model=kb_schemas.KnowledgeFactResponse, status_code=status.HTTP_201_CREATED) +def create_fact( + payload: kb_schemas.KnowledgeFactCreate, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + _get_topic(db, payload.topic_id) + if payload.category_id is not None: + cat = _get_category(db, payload.category_id) + if cat.topic_id != payload.topic_id: + raise HTTPException(status_code=400, detail="Category belongs to a different topic") + fact = kb_models.KnowledgeFact( + fact=payload.fact, + topic_id=payload.topic_id, + category_id=payload.category_id, + ) + db.add(fact) + db.commit() + db.refresh(fact) + return fact + + +@router.get("/knowledge-facts/{fact_id}", response_model=kb_schemas.KnowledgeFactResponse) +def get_fact( + fact_id: int, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + fact = db.query(kb_models.KnowledgeFact).filter(kb_models.KnowledgeFact.id == fact_id).first() + if not fact: + raise HTTPException(status_code=404, detail="Fact not found") + return fact + + +@router.patch("/knowledge-facts/{fact_id}", response_model=kb_schemas.KnowledgeFactResponse) +def update_fact( + fact_id: int, + payload: kb_schemas.KnowledgeFactUpdate, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + fact = db.query(kb_models.KnowledgeFact).filter(kb_models.KnowledgeFact.id == fact_id).first() + if not fact: + raise HTTPException(status_code=404, detail="Fact not found") + data = payload.model_dump(exclude_unset=True) + if "category_id" in data and data["category_id"] is not None: + cat = _get_category(db, data["category_id"]) + if cat.topic_id != fact.topic_id: + raise HTTPException(status_code=400, detail="Category belongs to a different topic") + for field, value in data.items(): + setattr(fact, field, value) + db.commit() + db.refresh(fact) + return fact + + +@router.delete("/knowledge-facts/{fact_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_fact( + fact_id: int, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + fact = db.query(kb_models.KnowledgeFact).filter(kb_models.KnowledgeFact.id == fact_id).first() + if not fact: + raise HTTPException(status_code=404, detail="Fact not found") + db.delete(fact) + db.commit() + return None + + +# =========================================================================== +# Project <-> KnowledgeBase links +# =========================================================================== +@router.get("/projects/{project_id}/knowledge-bases", response_model=List[kb_schemas.KnowledgeBaseResponse]) +def list_project_knowledge_bases( + project_id: str, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_READ) + project = _resolve_project(db, project_id) + linked_ids = [ + row.knowledge_base_id + for row in db.query(kb_models.ProjectKnowledgeBase.knowledge_base_id) + .filter(kb_models.ProjectKnowledgeBase.project_id == project.id) + .all() + ] + if not linked_ids: + return [] + return db.query(kb_models.KnowledgeBase).filter(kb_models.KnowledgeBase.id.in_(linked_ids)).all() + + +@router.post("/projects/{project_id}/knowledge-bases", response_model=kb_schemas.KnowledgeBaseResponse, status_code=status.HTTP_201_CREATED) +def link_knowledge_base_to_project( + project_id: str, + payload: kb_schemas.ProjectKnowledgeBaseLink, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + project = _resolve_project(db, project_id) + kb = _resolve_kb(db, payload.knowledge_base) + existing = ( + db.query(kb_models.ProjectKnowledgeBase) + .filter( + kb_models.ProjectKnowledgeBase.project_id == project.id, + kb_models.ProjectKnowledgeBase.knowledge_base_id == kb.id, + ) + .first() + ) + if not existing: + db.add(kb_models.ProjectKnowledgeBase(project_id=project.id, knowledge_base_id=kb.id)) + db.commit() + return kb + + +@router.delete("/projects/{project_id}/knowledge-bases/{kb_id}", status_code=status.HTTP_204_NO_CONTENT) +def unlink_knowledge_base_from_project( + project_id: str, + kb_id: str, + db: Session = Depends(get_db), + current_user: models.User = Depends(get_current_user_or_apikey), +): + _require_perm(db, current_user, PERM_UPDATE) + project = _resolve_project(db, project_id) + kb = _resolve_kb(db, kb_id) + db.query(kb_models.ProjectKnowledgeBase).filter( + kb_models.ProjectKnowledgeBase.project_id == project.id, + kb_models.ProjectKnowledgeBase.knowledge_base_id == kb.id, + ).delete(synchronize_session=False) + db.commit() + return None diff --git a/app/init_bootstrap.py b/app/init_bootstrap.py index f1ddcf0..1bae162 100644 --- a/app/init_bootstrap.py +++ b/app/init_bootstrap.py @@ -39,6 +39,11 @@ DEFAULT_PERMISSIONS = [ ("project.create", "Create a project", "project"), ("project.delete", "Delete project", "project"), ("project.manage_members", "Manage project members", "project"), + # Knowledge base permissions + ("knowledge-base.read", "View knowledge bases", "knowledge-base"), + ("knowledge-base.create", "Create a knowledge base", "knowledge-base"), + ("knowledge-base.update", "Edit a knowledge base and its structure", "knowledge-base"), + ("knowledge-base.delete", "Delete a knowledge base", "knowledge-base"), # Task/Milestone permissions ("task.create", "Create tasks", "task"), ("task.read", "View tasks", "task"), @@ -106,6 +111,7 @@ def init_default_permissions(db: Session) -> list[Permission]: # --------------------------------------------------------------------------- _MGR_PERMISSIONS = { "project.read", "project.write", "project.create", "project.manage_members", + "knowledge-base.read", "knowledge-base.create", "knowledge-base.update", "knowledge-base.delete", "task.create", "task.read", "task.write", "task.delete", "milestone.create", "milestone.read", "milestone.write", "milestone.delete", "milestone.freeze", "milestone.start", "milestone.close", @@ -118,6 +124,7 @@ _MGR_PERMISSIONS = { _DEV_PERMISSIONS = { "project.read", + "knowledge-base.read", "knowledge-base.update", "task.create", "task.read", "task.write", "milestone.read", "task.close", "task.reopen_closed", "task.reopen_completed", @@ -138,6 +145,7 @@ _ACCOUNT_MANAGER_PERMISSIONS = { # without admin intervention. _GENERAL_AGENT_PERMISSIONS = { "project.read", + "knowledge-base.read", "task.read", "milestone.read", "monitor.read", diff --git a/app/main.py b/app/main.py index 86173ae..3c393b0 100644 --- a/app/main.py +++ b/app/main.py @@ -79,6 +79,7 @@ from app.api.routers.schedule_type import router as schedule_type_router from app.api.routers.schedule_type_special_slot import router as schedule_type_special_slot_router from app.api.routers.calendar import router as calendar_router from app.api.routers.oidc import router as oidc_router +from app.api.routers.knowledge import router as knowledge_router app.include_router(auth_router) app.include_router(oidc_router) @@ -99,6 +100,7 @@ app.include_router(essentials_router) app.include_router(schedule_type_router) app.include_router(schedule_type_special_slot_router) app.include_router(calendar_router) +app.include_router(knowledge_router) # Auto schema migration for lightweight deployments @@ -488,7 +490,7 @@ def _sync_default_user_roles(db): @app.on_event("startup") def startup(): from app.core.config import Base, engine, SessionLocal - from app.models import models, webhook, apikey, activity, milestone, notification, worklog, monitor, role_permission, task, support, meeting, proposal, propose, essential, agent, calendar, minimum_workload, schedule_type, schedule_type_special_slot, oidc_settings + from app.models import models, webhook, apikey, activity, milestone, notification, worklog, monitor, role_permission, task, support, meeting, proposal, propose, essential, agent, calendar, minimum_workload, schedule_type, schedule_type_special_slot, oidc_settings, knowledge Base.metadata.create_all(bind=engine) _migrate_schema() diff --git a/app/models/knowledge.py b/app/models/knowledge.py new file mode 100644 index 0000000..2265fcf --- /dev/null +++ b/app/models/knowledge.py @@ -0,0 +1,101 @@ +"""Knowledge Base models. + +Mirrors the Project feature's shape (human-friendly *code*, creator FK, +created/updated timestamps). Hierarchy is: + + knowledge_base + └─ knowledge_topic (unique per (topic, knowledge_base_id)) + ├─ knowledge_fact (category_id NULL → fact lives on the topic) + └─ knowledge_category (parent NULL → top-level category in topic) + ├─ knowledge_fact + └─ knowledge_category (parent → nested) + +`project_knowledge_base` is the M2M link between projects and knowledge bases. + +Relationships are intentionally kept minimal (no ORM cascade on the +self-referential category tree); deletion ordering is handled explicitly in +the router to stay clear of FK-ordering surprises under MySQL. +""" +from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, UniqueConstraint +from sqlalchemy.orm import relationship +from sqlalchemy.sql import func + +from app.core.config import Base + + +class KnowledgeBase(Base): + __tablename__ = "knowledge_bases" + + id = Column(Integer, primary_key=True, index=True) + knowledge_base_code = Column(String(16), unique=True, index=True, nullable=True) + title = Column(String(200), nullable=False) + description = Column(Text, nullable=True) + created_by = Column(Integer, ForeignKey("users.id"), nullable=False) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + last_updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + creator = relationship("User", foreign_keys=[created_by]) + + +class KnowledgeTopic(Base): + __tablename__ = "knowledge_topics" + __table_args__ = ( + UniqueConstraint("topic", "knowledge_base_id", name="uq_knowledge_topic_kb"), + ) + + id = Column(Integer, primary_key=True, index=True) + topic = Column(String(200), nullable=False) + knowledge_base_id = Column(Integer, ForeignKey("knowledge_bases.id"), nullable=False, index=True) + description = Column(Text, nullable=True) + created_by = Column(Integer, ForeignKey("users.id"), nullable=False) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + last_updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + creator = relationship("User", foreign_keys=[created_by]) + + +class KnowledgeCategory(Base): + __tablename__ = "knowledge_categories" + __table_args__ = ( + # NOTE: MySQL treats NULLs as distinct in a UNIQUE index, so this only + # enforces uniqueness for non-NULL `parent`. Top-level categories + # (parent IS NULL) are de-duped in the router (application-level check). + UniqueConstraint("topic_id", "parent", "name", name="uq_knowledge_category_triple"), + ) + + id = Column(Integer, primary_key=True, index=True) + name = Column(String(200), nullable=False) + parent = Column(Integer, ForeignKey("knowledge_categories.id"), nullable=True, index=True) + topic_id = Column(Integer, ForeignKey("knowledge_topics.id"), nullable=False, index=True) + description = Column(Text, nullable=True) + created_by = Column(Integer, ForeignKey("users.id"), nullable=True) + last_updated_by = Column(Integer, ForeignKey("users.id"), nullable=True) + + +class KnowledgeFact(Base): + __tablename__ = "knowledge_facts" + + id = Column(Integer, primary_key=True, index=True) + category_id = Column(Integer, ForeignKey("knowledge_categories.id"), nullable=True, index=True) + topic_id = Column(Integer, ForeignKey("knowledge_topics.id"), nullable=False, index=True) + fact = Column(Text, nullable=False) + last_updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class ProjectKnowledgeBase(Base): + __tablename__ = "project_knowledge_bases" + __table_args__ = ( + UniqueConstraint("project_id", "knowledge_base_id", name="uq_project_knowledge_base"), + ) + + id = Column(Integer, primary_key=True, index=True) + project_id = Column(Integer, ForeignKey("projects.id"), nullable=False, index=True) + knowledge_base_id = Column(Integer, ForeignKey("knowledge_bases.id"), nullable=False, index=True) + + +class KnowledgeBaseCodeCounter(Base): + __tablename__ = "knowledge_base_code_counters" + + id = Column(Integer, primary_key=True, index=True) + prefix = Column(String(16), unique=True, index=True, nullable=False) + next_value = Column(Integer, default=0) diff --git a/app/schemas/knowledge.py b/app/schemas/knowledge.py new file mode 100644 index 0000000..59dbbae --- /dev/null +++ b/app/schemas/knowledge.py @@ -0,0 +1,166 @@ +"""Pydantic schemas for the Knowledge Base feature.""" +from pydantic import BaseModel +from typing import Optional, List +from datetime import datetime + + +# -------------------------------------------------------------------------- +# Knowledge Base +# -------------------------------------------------------------------------- +class KnowledgeBaseBase(BaseModel): + title: str + description: Optional[str] = None + + +class KnowledgeBaseCreate(KnowledgeBaseBase): + pass + + +class KnowledgeBaseUpdate(BaseModel): + title: Optional[str] = None + description: Optional[str] = None + + +class KnowledgeBaseResponse(BaseModel): + id: int + knowledge_base_code: Optional[str] = None + title: str + description: Optional[str] = None + created_by: int + created_at: Optional[datetime] = None + last_updated_at: Optional[datetime] = None + + class Config: + from_attributes = True + + +# -------------------------------------------------------------------------- +# Topic +# -------------------------------------------------------------------------- +class KnowledgeTopicBase(BaseModel): + topic: str + description: Optional[str] = None + + +class KnowledgeTopicCreate(KnowledgeTopicBase): + pass + + +class KnowledgeTopicUpdate(BaseModel): + topic: Optional[str] = None + description: Optional[str] = None + + +class KnowledgeTopicResponse(BaseModel): + id: int + topic: str + knowledge_base_id: int + description: Optional[str] = None + created_by: int + created_at: Optional[datetime] = None + last_updated_at: Optional[datetime] = None + + class Config: + from_attributes = True + + +# -------------------------------------------------------------------------- +# Category +# -------------------------------------------------------------------------- +class KnowledgeCategoryBase(BaseModel): + name: str + description: Optional[str] = None + + +class KnowledgeCategoryCreate(KnowledgeCategoryBase): + topic_id: int + parent: Optional[int] = None + + +class KnowledgeCategoryUpdate(BaseModel): + name: Optional[str] = None + description: Optional[str] = None + parent: Optional[int] = None + + +class KnowledgeCategoryResponse(BaseModel): + id: int + name: str + parent: Optional[int] = None + topic_id: int + description: Optional[str] = None + created_by: Optional[int] = None + last_updated_by: Optional[int] = None + + class Config: + from_attributes = True + + +# -------------------------------------------------------------------------- +# Fact +# -------------------------------------------------------------------------- +class KnowledgeFactBase(BaseModel): + fact: str + + +class KnowledgeFactCreate(KnowledgeFactBase): + topic_id: int + category_id: Optional[int] = None + + +class KnowledgeFactUpdate(BaseModel): + fact: Optional[str] = None + category_id: Optional[int] = None + + +class KnowledgeFactResponse(BaseModel): + id: int + category_id: Optional[int] = None + topic_id: int + fact: str + last_updated_at: Optional[datetime] = None + + class Config: + from_attributes = True + + +# -------------------------------------------------------------------------- +# Project <-> KnowledgeBase link +# -------------------------------------------------------------------------- +class ProjectKnowledgeBaseLink(BaseModel): + # Accept either a numeric id or a knowledge_base_code (mirrors how + # projects are referenced elsewhere). + knowledge_base: str + + +# -------------------------------------------------------------------------- +# Nested tree (read-only aggregate) +# -------------------------------------------------------------------------- +class CategoryTreeNode(BaseModel): + id: int + name: str + parent: Optional[int] = None + topic_id: int + description: Optional[str] = None + categories: List["CategoryTreeNode"] = [] + facts: List[KnowledgeFactResponse] = [] + + +class TopicTreeNode(BaseModel): + id: int + topic: str + knowledge_base_id: int + description: Optional[str] = None + categories: List[CategoryTreeNode] = [] + facts: List[KnowledgeFactResponse] = [] + + +class KnowledgeBaseTree(BaseModel): + id: int + knowledge_base_code: Optional[str] = None + title: str + description: Optional[str] = None + topics: List[TopicTreeNode] = [] + + +CategoryTreeNode.model_rebuild() diff --git a/tests/test_knowledge_base.py b/tests/test_knowledge_base.py new file mode 100644 index 0000000..2ca7803 --- /dev/null +++ b/tests/test_knowledge_base.py @@ -0,0 +1,147 @@ +"""Knowledge Base API tests — CRUD, hierarchy, uniqueness, tree, links, RBAC.""" +from tests.conftest import auth_header + + +def _create_kb(client, token, title="Infra Runbook", description="ops notes"): + r = client.post( + "/knowledge-bases", + json={"title": title, "description": description}, + headers=auth_header(token), + ) + assert r.status_code == 201, r.text + return r.json() + + +class TestKnowledgeBaseCRUD: + def test_create_generates_code(self, client, seed): + kb = _create_kb(client, seed["admin_token"], title="Infra Runbook") + assert kb["title"] == "Infra Runbook" + assert kb["knowledge_base_code"] # auto-generated, non-empty + assert kb["created_by"] == seed["admin_user"].id + + def test_create_requires_permission(self, client, seed): + # dev role has no knowledge-base.create + r = client.post( + "/knowledge-bases", + json={"title": "Nope"}, + headers=auth_header(seed["dev_token"]), + ) + assert r.status_code == 403 + + def test_get_by_id_and_code(self, client, seed): + kb = _create_kb(client, seed["admin_token"]) + by_id = client.get(f"/knowledge-bases/{kb['id']}", headers=auth_header(seed["admin_token"])) + by_code = client.get(f"/knowledge-bases/{kb['knowledge_base_code']}", headers=auth_header(seed["admin_token"])) + assert by_id.status_code == 200 and by_code.status_code == 200 + assert by_id.json()["id"] == by_code.json()["id"] == kb["id"] + + def test_update_and_list(self, client, seed): + kb = _create_kb(client, seed["admin_token"]) + r = client.patch( + f"/knowledge-bases/{kb['id']}", + json={"description": "updated"}, + headers=auth_header(seed["admin_token"]), + ) + assert r.status_code == 200 and r.json()["description"] == "updated" + lst = client.get("/knowledge-bases", headers=auth_header(seed["admin_token"])) + assert lst.status_code == 200 and any(k["id"] == kb["id"] for k in lst.json()) + + def test_delete_cascades(self, client, seed): + token = seed["admin_token"] + kb = _create_kb(client, token) + topic = client.post(f"/knowledge-bases/{kb['id']}/topics", json={"topic": "Net"}, headers=auth_header(token)).json() + client.post("/knowledge-facts", json={"topic_id": topic["id"], "fact": "x"}, headers=auth_header(token)) + r = client.delete(f"/knowledge-bases/{kb['id']}", headers=auth_header(token)) + assert r.status_code == 204 + assert client.get(f"/knowledge-bases/{kb['id']}", headers=auth_header(token)).status_code == 404 + assert client.get(f"/knowledge-topics/{topic['id']}", headers=auth_header(token)).status_code == 404 + + +class TestHierarchy: + def test_topic_unique_per_kb(self, client, seed): + token = seed["admin_token"] + kb = _create_kb(client, token) + r1 = client.post(f"/knowledge-bases/{kb['id']}/topics", json={"topic": "Routing"}, headers=auth_header(token)) + r2 = client.post(f"/knowledge-bases/{kb['id']}/topics", json={"topic": "Routing"}, headers=auth_header(token)) + assert r1.status_code == 201 and r2.status_code == 400 + + def test_category_triple_unique_and_nesting(self, client, seed): + token = seed["admin_token"] + kb = _create_kb(client, token) + topic = client.post(f"/knowledge-bases/{kb['id']}/topics", json={"topic": "T"}, headers=auth_header(token)).json() + # top-level category + c1 = client.post("/knowledge-categories", json={"topic_id": topic["id"], "name": "DNS"}, headers=auth_header(token)) + assert c1.status_code == 201 + # duplicate top-level (parent NULL) rejected at app level + dup = client.post("/knowledge-categories", json={"topic_id": topic["id"], "name": "DNS"}, headers=auth_header(token)) + assert dup.status_code == 400 + # nested category with same name under different parent is allowed + child = client.post( + "/knowledge-categories", + json={"topic_id": topic["id"], "name": "DNS", "parent": c1.json()["id"]}, + headers=auth_header(token), + ) + assert child.status_code == 201 + + def test_no_cycle_on_reparent(self, client, seed): + token = seed["admin_token"] + kb = _create_kb(client, token) + topic = client.post(f"/knowledge-bases/{kb['id']}/topics", json={"topic": "T"}, headers=auth_header(token)).json() + a = client.post("/knowledge-categories", json={"topic_id": topic["id"], "name": "A"}, headers=auth_header(token)).json() + b = client.post("/knowledge-categories", json={"topic_id": topic["id"], "name": "B", "parent": a["id"]}, headers=auth_header(token)).json() + # try to move A under its descendant B -> rejected + r = client.patch(f"/knowledge-categories/{a['id']}", json={"parent": b["id"]}, headers=auth_header(token)) + assert r.status_code == 400 + + def test_tree_shape(self, client, seed): + token = seed["admin_token"] + kb = _create_kb(client, token) + topic = client.post(f"/knowledge-bases/{kb['id']}/topics", json={"topic": "T"}, headers=auth_header(token)).json() + cat = client.post("/knowledge-categories", json={"topic_id": topic["id"], "name": "C"}, headers=auth_header(token)).json() + # fact directly on topic + client.post("/knowledge-facts", json={"topic_id": topic["id"], "fact": "topic-fact"}, headers=auth_header(token)) + # fact under category + client.post("/knowledge-facts", json={"topic_id": topic["id"], "category_id": cat["id"], "fact": "cat-fact"}, headers=auth_header(token)) + + tree = client.get(f"/knowledge-bases/{kb['id']}/tree", headers=auth_header(token)).json() + assert len(tree["topics"]) == 1 + t = tree["topics"][0] + assert [f["fact"] for f in t["facts"]] == ["topic-fact"] + assert len(t["categories"]) == 1 + assert [f["fact"] for f in t["categories"][0]["facts"]] == ["cat-fact"] + + +class TestProjectLinks: + def test_link_unlink(self, client, seed): + token = seed["admin_token"] + kb = _create_kb(client, token) + # link by code + r = client.post( + "/projects/TPRJ/knowledge-bases", + json={"knowledge_base": kb["knowledge_base_code"]}, + headers=auth_header(token), + ) + assert r.status_code == 201 + linked = client.get("/projects/TPRJ/knowledge-bases", headers=auth_header(token)).json() + assert any(k["id"] == kb["id"] for k in linked) + # filter list by project + filtered = client.get(f"/knowledge-bases?project=TPRJ", headers=auth_header(token)).json() + assert any(k["id"] == kb["id"] for k in filtered) + # unlink + r = client.delete(f"/projects/TPRJ/knowledge-bases/{kb['id']}", headers=auth_header(token)) + assert r.status_code == 204 + linked = client.get("/projects/TPRJ/knowledge-bases", headers=auth_header(token)).json() + assert not any(k["id"] == kb["id"] for k in linked) + + def test_link_is_idempotent(self, client, seed): + token = seed["admin_token"] + kb = _create_kb(client, token) + for _ in range(2): + r = client.post( + "/projects/TPRJ/knowledge-bases", + json={"knowledge_base": str(kb["id"])}, + headers=auth_header(token), + ) + assert r.status_code == 201 + linked = client.get("/projects/TPRJ/knowledge-bases", headers=auth_header(token)).json() + assert sum(1 for k in linked if k["id"] == kb["id"]) == 1