fix(knowledge): use EntityNotFoundError/DuplicateEntityError instead of DomainError(status_code=)
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled

This commit is contained in:
kitos
2026-05-20 15:21:36 +02:00
parent 9546ef8bc8
commit 4fba4152d9
3 changed files with 14 additions and 17 deletions

View File

@@ -124,11 +124,8 @@ def get_playbook_by_technique_type(
):
pb = pb_svc.get_playbook_by_technique_type(db, technique_id, playbook_type)
if not pb:
from app.domain.errors import DomainError
raise DomainError(
f"No '{playbook_type}' playbook for technique {technique_id}",
status_code=404,
)
from app.domain.errors import EntityNotFoundError
raise EntityNotFoundError("Playbook", f"{technique_id}/{playbook_type}")
return pb

View File

@@ -6,7 +6,7 @@ from uuid import UUID
from sqlalchemy.orm import Session
from app.domain.errors import DomainError
from app.domain.errors import EntityNotFoundError
from app.models.knowledge import LessonLearned
@@ -16,7 +16,7 @@ def _get_or_404(db: Session, ll_id: UUID) -> LessonLearned:
LessonLearned.is_active == True,
).first()
if not ll:
raise DomainError(f"Lesson Learned {ll_id} not found", status_code=404)
raise EntityNotFoundError("LessonLearned", str(ll_id))
return ll

View File

@@ -6,7 +6,9 @@ from uuid import UUID
from sqlalchemy.orm import Session
from app.domain.errors import DomainError
from app.domain.errors import (
DomainError, EntityNotFoundError, DuplicateEntityError, BusinessRuleViolation
)
from app.models.knowledge import Playbook, PlaybookVersion
from app.models.technique import Technique
@@ -17,7 +19,7 @@ def _get_or_404(db: Session, playbook_id: UUID) -> Playbook:
Playbook.is_active == True,
).first()
if not pb:
raise DomainError(f"Playbook {playbook_id} not found", status_code=404)
raise EntityNotFoundError("Playbook", str(playbook_id))
return pb
@@ -88,7 +90,7 @@ def create_playbook(db: Session, data: dict, user_id: UUID) -> Playbook:
# Validate technique exists
tech = db.query(Technique).filter(Technique.id == technique_id).first()
if not tech:
raise DomainError(f"Technique {technique_id} not found", status_code=404)
raise EntityNotFoundError("Technique", str(technique_id))
# Check for existing (even soft-deleted) to reactivate
existing = db.query(Playbook).filter(
@@ -98,10 +100,10 @@ def create_playbook(db: Session, data: dict, user_id: UUID) -> Playbook:
if existing:
if existing.is_active:
raise DomainError(
f"Playbook ({playbook_type}) for technique {technique_id} already exists. "
"Use PATCH to update it.",
status_code=409,
raise DuplicateEntityError(
"Playbook",
"technique_id+playbook_type",
f"{technique_id}/{playbook_type}",
)
# Reactivate soft-deleted playbook
existing.is_active = True
@@ -170,9 +172,7 @@ def restore_version(db: Session, playbook_id: UUID, version_number: int, user_id
PlaybookVersion.version == version_number,
).first()
if not snap:
raise DomainError(
f"Version {version_number} not found for playbook {playbook_id}", status_code=404
)
raise EntityNotFoundError("PlaybookVersion", f"{playbook_id}/v{version_number}")
# Snapshot current state before restoring
_snapshot(db, pb, user_id, f"Auto-snapshot before restore to v{version_number}")