breakpilot-compliance/backend-compliance/compliance/services/vendor_compliance_extra_service.py

# mypy: disable-error-code="arg-type,assignment,union-attr,no-any-return,attr-defined,index,call-overload,type-arg,var-annotated,misc,call-arg,return-value"
"""
Vendor compliance extra entities — Findings, Control Instances, and
Controls Library CRUD.

Phase 1 Step 4: extracted from ``compliance.api.vendor_compliance_routes``.
Shares helpers with ``compliance.services.vendor_compliance_service`` and
row converters from ``compliance.services.vendor_compliance_sub_service``.
"""

import json
import uuid
from datetime import datetime, timezone
from typing import Optional

from sqlalchemy import text
from sqlalchemy.orm import Session

from compliance.domain import NotFoundError
from compliance.services.vendor_compliance_service import (
    DEFAULT_TENANT_ID,
    _get,
    _ok,
    _to_snake,
    _ts,
)
from compliance.services.vendor_compliance_sub_service import (
    _control_instance_to_response,
    _finding_to_response,
)


# ============================================================================
# FindingService
# ============================================================================


class FindingService:
    """Vendor findings CRUD."""

    def __init__(self, db: Session) -> None:
        self._db = db

    def list_findings(
        self,
        tenant_id: Optional[str] = None,
        vendor_id: Optional[str] = None,
        severity: Optional[str] = None,
        status: Optional[str] = None,
        skip: int = 0,
        limit: int = 100,
    ) -> dict:
        tid = tenant_id or DEFAULT_TENANT_ID
        where = ["tenant_id = :tid"]
        params: dict = {"tid": tid}
        if vendor_id:
            where.append("vendor_id = :vendor_id")
            params["vendor_id"] = vendor_id
        if severity:
            where.append("severity = :severity")
            params["severity"] = severity
        if status:
            where.append("status = :status")
            params["status"] = status
        where_clause = " AND ".join(where)
        params["lim"] = limit
        params["off"] = skip

        rows = self._db.execute(text(f"""
            SELECT * FROM vendor_findings
            WHERE {where_clause}
            ORDER BY created_at DESC
            LIMIT :lim OFFSET :off
        """), params).fetchall()
        return _ok([_finding_to_response(r) for r in rows])

    def get_finding(self, finding_id: str) -> dict:
        row = self._db.execute(
            text("SELECT * FROM vendor_findings WHERE id = :id"),
            {"id": finding_id},
        ).fetchone()
        if not row:
            raise NotFoundError("Finding not found")
        return _ok(_finding_to_response(row))

    def create_finding(self, body: dict) -> dict:
        data = _to_snake(body)
        fid = str(uuid.uuid4())
        tid = data.get("tenant_id", DEFAULT_TENANT_ID)
        now = datetime.now(timezone.utc).isoformat()

        self._db.execute(text("""
            INSERT INTO vendor_findings (
                id, tenant_id, vendor_id, contract_id,
                finding_type, category, severity,
                title, description, recommendation,
                citations, status, assignee, due_date,
                created_at, updated_at, created_by
            ) VALUES (
                :id, :tenant_id, :vendor_id, :contract_id,
                :finding_type, :category, :severity,
                :title, :description, :recommendation,
                CAST(:citations AS jsonb), :status, :assignee, :due_date,
                :created_at, :updated_at, :created_by
            )
        """), {
            "id": fid, "tenant_id": tid,
            "vendor_id": data.get("vendor_id", ""),
            "contract_id": data.get("contract_id"),
            "finding_type": data.get("finding_type", "UNKNOWN"),
            "category": data.get("category", ""),
            "severity": data.get("severity", "MEDIUM"),
            "title": data.get("title", ""),
            "description": data.get("description", ""),
            "recommendation": data.get("recommendation", ""),
            "citations": json.dumps(data.get("citations", [])),
            "status": data.get("status", "OPEN"),
            "assignee": data.get("assignee", ""),
            "due_date": data.get("due_date"),
            "created_at": now, "updated_at": now,
            "created_by": data.get("created_by", "system"),
        })
        self._db.commit()
        row = self._db.execute(
            text("SELECT * FROM vendor_findings WHERE id = :id"),
            {"id": fid},
        ).fetchone()
        return _ok(_finding_to_response(row))

    def update_finding(self, finding_id: str, body: dict) -> dict:
        existing = self._db.execute(
            text("SELECT id FROM vendor_findings WHERE id = :id"),
            {"id": finding_id},
        ).fetchone()
        if not existing:
            raise NotFoundError("Finding not found")

        data = _to_snake(body)
        now = datetime.now(timezone.utc).isoformat()
        allowed = [
            "vendor_id", "contract_id", "finding_type", "category",
            "severity", "title", "description", "recommendation",
            "status", "assignee", "due_date",
            "resolution", "resolved_at", "resolved_by",
        ]
        jsonb_fields = ["citations"]

        sets = ["updated_at = :updated_at"]
        params: dict = {"id": finding_id, "updated_at": now}
        for col in allowed:
            if col in data:
                sets.append(f"{col} = :{col}")
                params[col] = data[col]
        for col in jsonb_fields:
            if col in data:
                sets.append(f"{col} = CAST(:{col} AS jsonb)")
                params[col] = json.dumps(data[col])

        self._db.execute(
            text(
                f"UPDATE vendor_findings SET {', '.join(sets)} WHERE id = :id",
            ),
            params,
        )
        self._db.commit()
        row = self._db.execute(
            text("SELECT * FROM vendor_findings WHERE id = :id"),
            {"id": finding_id},
        ).fetchone()
        return _ok(_finding_to_response(row))

    def delete_finding(self, finding_id: str) -> dict:
        result = self._db.execute(
            text("DELETE FROM vendor_findings WHERE id = :id"),
            {"id": finding_id},
        )
        self._db.commit()
        if result.rowcount == 0:
            raise NotFoundError("Finding not found")
        return _ok({"deleted": True})


# ============================================================================
# ControlInstanceService
# ============================================================================


class ControlInstanceService:
    """Vendor control instances CRUD."""

    def __init__(self, db: Session) -> None:
        self._db = db

    def list_instances(
        self,
        tenant_id: Optional[str] = None,
        vendor_id: Optional[str] = None,
        skip: int = 0,
        limit: int = 100,
    ) -> dict:
        tid = tenant_id or DEFAULT_TENANT_ID
        where = ["tenant_id = :tid"]
        params: dict = {"tid": tid}
        if vendor_id:
            where.append("vendor_id = :vendor_id")
            params["vendor_id"] = vendor_id
        where_clause = " AND ".join(where)
        params["lim"] = limit
        params["off"] = skip

        rows = self._db.execute(text(f"""
            SELECT * FROM vendor_control_instances
            WHERE {where_clause}
            ORDER BY created_at DESC
            LIMIT :lim OFFSET :off
        """), params).fetchall()
        return _ok([_control_instance_to_response(r) for r in rows])

    def get_instance(self, instance_id: str) -> dict:
        row = self._db.execute(
            text("SELECT * FROM vendor_control_instances WHERE id = :id"),
            {"id": instance_id},
        ).fetchone()
        if not row:
            raise NotFoundError("Control instance not found")
        return _ok(_control_instance_to_response(row))

    def create_instance(self, body: dict) -> dict:
        data = _to_snake(body)
        ciid = str(uuid.uuid4())
        tid = data.get("tenant_id", DEFAULT_TENANT_ID)
        now = datetime.now(timezone.utc).isoformat()

        self._db.execute(text("""
            INSERT INTO vendor_control_instances (
                id, tenant_id, vendor_id, control_id, control_domain,
                status, evidence_ids, notes,
                last_assessed_at, last_assessed_by, next_assessment_date,
                created_at, updated_at, created_by
            ) VALUES (
                :id, :tenant_id, :vendor_id, :control_id, :control_domain,
                :status, CAST(:evidence_ids AS jsonb), :notes,
                :last_assessed_at, :last_assessed_by,
                :next_assessment_date,
                :created_at, :updated_at, :created_by
            )
        """), {
            "id": ciid, "tenant_id": tid,
            "vendor_id": data.get("vendor_id", ""),
            "control_id": data.get("control_id", ""),
            "control_domain": data.get("control_domain", ""),
            "status": data.get("status", "PLANNED"),
            "evidence_ids": json.dumps(data.get("evidence_ids", [])),
            "notes": data.get("notes", ""),
            "last_assessed_at": data.get("last_assessed_at"),
            "last_assessed_by": data.get("last_assessed_by", ""),
            "next_assessment_date": data.get("next_assessment_date"),
            "created_at": now, "updated_at": now,
            "created_by": data.get("created_by", "system"),
        })
        self._db.commit()
        row = self._db.execute(
            text("SELECT * FROM vendor_control_instances WHERE id = :id"),
            {"id": ciid},
        ).fetchone()
        return _ok(_control_instance_to_response(row))

    def update_instance(self, instance_id: str, body: dict) -> dict:
        existing = self._db.execute(
            text("SELECT id FROM vendor_control_instances WHERE id = :id"),
            {"id": instance_id},
        ).fetchone()
        if not existing:
            raise NotFoundError("Control instance not found")

        data = _to_snake(body)
        now = datetime.now(timezone.utc).isoformat()
        allowed = [
            "vendor_id", "control_id", "control_domain",
            "status", "notes",
            "last_assessed_at", "last_assessed_by",
            "next_assessment_date",
        ]
        jsonb_fields = ["evidence_ids"]

        sets = ["updated_at = :updated_at"]
        params: dict = {"id": instance_id, "updated_at": now}
        for col in allowed:
            if col in data:
                sets.append(f"{col} = :{col}")
                params[col] = data[col]
        for col in jsonb_fields:
            if col in data:
                sets.append(f"{col} = CAST(:{col} AS jsonb)")
                params[col] = json.dumps(data[col])

        self._db.execute(text(
            f"UPDATE vendor_control_instances SET {', '.join(sets)} "
            f"WHERE id = :id",
        ), params)
        self._db.commit()
        row = self._db.execute(
            text("SELECT * FROM vendor_control_instances WHERE id = :id"),
            {"id": instance_id},
        ).fetchone()
        return _ok(_control_instance_to_response(row))

    def delete_instance(self, instance_id: str) -> dict:
        result = self._db.execute(
            text("DELETE FROM vendor_control_instances WHERE id = :id"),
            {"id": instance_id},
        )
        self._db.commit()
        if result.rowcount == 0:
            raise NotFoundError("Control instance not found")
        return _ok({"deleted": True})


# ============================================================================
# ControlsLibraryService
# ============================================================================


class ControlsLibraryService:
    """Controls library (vendor_compliance_controls catalog)."""

    def __init__(self, db: Session) -> None:
        self._db = db

    def list_controls(
        self,
        tenant_id: Optional[str] = None,
        domain: Optional[str] = None,
    ) -> dict:
        tid = tenant_id or DEFAULT_TENANT_ID
        where = ["tenant_id = :tid"]
        params: dict = {"tid": tid}
        if domain:
            where.append("domain = :domain")
            params["domain"] = domain
        where_clause = " AND ".join(where)

        rows = self._db.execute(text(f"""
            SELECT * FROM vendor_compliance_controls
            WHERE {where_clause}
            ORDER BY domain, control_code
        """), params).fetchall()

        items = []
        for r in rows:
            items.append({
                "id": str(r["id"]),
                "tenantId": r["tenant_id"],
                "domain": _get(r, "domain", ""),
                "controlCode": _get(r, "control_code", ""),
                "title": _get(r, "title", ""),
                "description": _get(r, "description", ""),
                "createdAt": _ts(r["created_at"]),
            })
        return _ok(items)

    def create_control(self, body: dict) -> dict:
        cid = str(uuid.uuid4())
        tid = body.get(
            "tenantId", body.get("tenant_id", DEFAULT_TENANT_ID),
        )
        now = datetime.now(timezone.utc).isoformat()

        self._db.execute(text("""
            INSERT INTO vendor_compliance_controls (
                id, tenant_id, domain, control_code, title, description,
                created_at
            ) VALUES (
                :id, :tenant_id, :domain, :control_code, :title,
                :description, :created_at
            )
        """), {
            "id": cid, "tenant_id": tid,
            "domain": body.get("domain", ""),
            "control_code": body.get(
                "controlCode", body.get("control_code", ""),
            ),
            "title": body.get("title", ""),
            "description": body.get("description", ""),
            "created_at": now,
        })
        self._db.commit()

        return _ok({
            "id": cid, "tenantId": tid,
            "domain": body.get("domain", ""),
            "controlCode": body.get(
                "controlCode", body.get("control_code", ""),
            ),
            "title": body.get("title", ""),
            "description": body.get("description", ""),
            "createdAt": now,
        })

    def delete_control(self, control_id: str) -> dict:
        result = self._db.execute(
            text("DELETE FROM vendor_compliance_controls WHERE id = :id"),
            {"id": control_id},
        )
        self._db.commit()
        if result.rowcount == 0:
            raise NotFoundError("Control not found")
        return _ok({"deleted": True})