fix: allow direct knowledge deletion

This commit is contained in:
2026-07-17 15:34:23 +08:00
parent 821cb26930
commit cadf44e0b8
3 changed files with 35 additions and 12 deletions

View File

@@ -192,12 +192,13 @@ def delete_knowledge(
) -> dict:
_require_super_admin(current_admin)
knowledge = _get_knowledge(db, knowledge_id)
if knowledge.lifecycle_status != "archived" or knowledge.status != 0:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="物理删除前必须先关闭并归档知识库",
)
# 清理早期按用户授权口径留下的历史关联,避免外键阻止知识库删除。
# 删除操作本身会立即停止开放并清理正式内容,不再要求管理员先执行“关闭、归档”两步。
knowledge.status = 0
knowledge.lifecycle_status = "archived"
db.add(knowledge)
db.flush()
# 清理按用户授权口径留下的历史关联,避免外键阻止知识库删除。
permissions = db.scalars(
select(UserKnowledgePermission).where(UserKnowledgePermission.knowledge_id == knowledge.id)
).all()
@@ -212,9 +213,16 @@ def delete_knowledge(
db.flush()
for row in db.scalars(select(KnowledgeSourceSnapshot).where(KnowledgeSourceSnapshot.knowledge_id == knowledge.id)).all():
db.delete(row)
db.delete(knowledge)
OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="delete", target_id=knowledge.id)
db.commit()
db.delete(knowledge)
try:
db.commit()
except IntegrityError as exc:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="知识库仍有关联数据,删除未执行,请刷新后重试",
) from exc
return api_success()

View File

@@ -10,6 +10,7 @@ from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from app.api.admin_knowledge import delete_knowledge
from app.api.admin_knowledge_lifecycle import batch_sync, batch_update_metadata, update_metadata
from app.models import Base
from app.models.admin import Admin, Role
@@ -134,6 +135,20 @@ def test_batch_metadata_update_reports_missing_items():
assert db.scalar(select(Knowledge).where(Knowledge.id == item.id)).knowledge_type == "qa"
def test_active_knowledge_can_be_deleted_without_manual_archive_steps():
with _database() as db:
item = _knowledge(db)
knowledge_id = item.id
response = delete_knowledge(knowledge_id, db, _admin())
assert response["code"] == 0
assert db.get(Knowledge, knowledge_id) is None
assert db.scalars(select(KnowledgeVersion).where(KnowledgeVersion.knowledge_id == knowledge_id)).all() == []
assert db.scalars(select(KnowledgeManifest).where(KnowledgeManifest.knowledge_id == knowledge_id)).all() == []
assert db.scalars(select(KnowledgeSourceSnapshot).where(KnowledgeSourceSnapshot.knowledge_id == knowledge_id)).all() == []
def test_batch_sync_continues_after_an_item_fails_and_reports_each_result():
with _database() as db:
first = _knowledge(db, 1)