"""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