hzhang
9feff8e008
New entities mirroring the Project shape: - knowledge_bases (human code, title, description, created_by, timestamps) - knowledge_topics (UNIQUE(topic, knowledge_base_id)) - knowledge_categories (self-referential parent; UNIQUE(topic_id, parent, name), with an app-level check for the NULL-parent case MySQL can't enforce) - knowledge_facts (category_id NULL → fact lives directly on the topic) - project_knowledge_bases (M2M project ↔ knowledge base) Adds full CRUD for KB/topic/category/fact, a nested /tree aggregate, project link/unlink/list, KB-code generation (same algorithm as project codes), and category cycle-prevention. Four global permissions (knowledge-base.create/read/update/delete) seeded in init_bootstrap and granted to admin/mgr/dev/general-agent/guest as appropriate. New tables auto-create via Base.metadata.create_all; router wired in main.py. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
736 lines
27 KiB
Python
736 lines
27 KiB
Python
"""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
|