feat: move all remaining inline logic from routers to services (Tier 2)

This commit is contained in:
2026-02-20 14:34:24 +01:00
parent 9e22fde746
commit 339d669498
17 changed files with 632 additions and 414 deletions

View File

@@ -0,0 +1,82 @@
"""D3FEND query service — framework-agnostic queries for defensive techniques."""
from __future__ import annotations
from typing import Optional
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.domain.errors import EntityNotFoundError
from app.models.defensive_technique import DefensiveTechnique
from app.models.technique import Technique
from app.services.d3fend_import_service import get_defenses_for_technique
from app.utils import escape_like
def list_defensive_techniques(
db: Session,
*,
tactic: Optional[str] = None,
search: Optional[str] = None,
offset: int = 0,
limit: int = 50,
) -> dict:
"""List D3FEND defensive techniques with optional filters."""
query = db.query(DefensiveTechnique)
if tactic:
query = query.filter(DefensiveTechnique.tactic == tactic)
if search:
pattern = f"%{escape_like(search)}%"
query = query.filter(
DefensiveTechnique.name.ilike(pattern)
| DefensiveTechnique.d3fend_id.ilike(pattern)
)
total = query.count()
items = query.order_by(DefensiveTechnique.d3fend_id).offset(offset).limit(limit).all()
return {
"total": total,
"offset": offset,
"limit": limit,
"items": [
{
"id": str(dt.id),
"d3fend_id": dt.d3fend_id,
"name": dt.name,
"description": dt.description,
"tactic": dt.tactic,
"d3fend_url": dt.d3fend_url,
}
for dt in items
],
}
def list_d3fend_tactics(db: Session) -> list[dict]:
"""Return a list of all D3FEND tactics with counts."""
rows = (
db.query(DefensiveTechnique.tactic, func.count(DefensiveTechnique.id))
.group_by(DefensiveTechnique.tactic)
.order_by(DefensiveTechnique.tactic)
.all()
)
return [{"tactic": tactic or "Unknown", "count": count} for tactic, count in rows]
def get_defenses_for_attack_technique(db: Session, mitre_id: str) -> dict:
"""Get all D3FEND defensive techniques mapped to a given ATT&CK technique."""
technique = db.query(Technique).filter(Technique.mitre_id == mitre_id).first()
if technique is None:
raise EntityNotFoundError("Technique", mitre_id)
defenses = get_defenses_for_technique(db, technique.id)
return {
"mitre_id": mitre_id,
"technique_name": technique.name,
"defenses": defenses,
"total": len(defenses),
}