company-ai-platform/app/modules/approvals/service.py

import json
import uuid

from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.core.time import utc_now
from app.modules.approvals.constants import ApprovalActionValue, ApprovalStatus
from app.modules.approvals.models import ApprovalRequest
from app.modules.approvals.schemas import ApprovalCreate
from app.modules.audit.constants import AuditAction, AuditRiskLevel, AuditSource
from app.modules.audit.schemas import AuditLogCreate
from app.modules.audit.service import AuditService


class ApprovalService:
    """Create, decide, and validate approval tickets for guarded actions."""

    def __init__(self, db: Session):
        self.db = db
        self.audit = AuditService(db)

    def create(self, payload: ApprovalCreate, applicant: str) -> ApprovalRequest:
        ticket = ApprovalRequest(
            ticket_id=f"APR-{uuid.uuid4().hex[:12].upper()}",
            domain=payload.domain,
            record_id=payload.record_id,
            action=payload.action,
            applicant=applicant,
            reason=payload.reason,
            payload=json.dumps(payload.payload, ensure_ascii=False, default=str),
        )
        self.db.add(ticket)
        self.db.commit()
        self.db.refresh(ticket)
        self.audit.log(
            AuditLogCreate(
                actor=applicant,
                source=AuditSource.APPROVAL,
                action=AuditAction.APPROVAL_CREATE,
                target_type=payload.domain,
                target_id=payload.record_id,
                risk_level=AuditRiskLevel.MEDIUM,
                request_payload=payload.model_dump(),
                response_payload={"ticket_id": ticket.ticket_id, "status": ticket.status},
            )
        )
        return ticket

    def list(self, status_filter: str | None = None, limit: int = 100) -> list[ApprovalRequest]:
        stmt = select(ApprovalRequest).order_by(ApprovalRequest.id.desc()).limit(min(limit, 500))
        if status_filter:
            stmt = stmt.where(ApprovalRequest.status == status_filter)
        return list(self.db.execute(stmt).scalars())

    def get_by_ticket(self, ticket_id: str) -> ApprovalRequest:
        ticket = self.db.execute(
            select(ApprovalRequest).where(ApprovalRequest.ticket_id == ticket_id)
        ).scalar_one_or_none()
        if ticket is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Approval ticket not found",
            )
        return ticket

    def decide(
        self,
        ticket_id: str,
        approver: str,
        approved: bool,
        comment: str | None,
    ) -> ApprovalRequest:
        ticket = self.get_by_ticket(ticket_id)
        if ticket.status != ApprovalStatus.PENDING:
            raise HTTPException(status_code=409, detail="Approval ticket already decided")
        ticket.status = ApprovalStatus.APPROVED if approved else ApprovalStatus.REJECTED
        ticket.approver = approver
        ticket.decision_comment = comment
        ticket.decided_at = utc_now()
        self.db.commit()
        self.db.refresh(ticket)
        self.audit.log(
            AuditLogCreate(
                actor=approver,
                source=AuditSource.APPROVAL,
                action=AuditAction.APPROVAL_APPROVE if approved else AuditAction.APPROVAL_REJECT,
                target_type=ticket.domain,
                target_id=ticket.record_id,
                risk_level=AuditRiskLevel.HIGH,
                request_payload={"ticket_id": ticket_id, "comment": comment},
                response_payload={"status": ticket.status},
            )
        )
        return ticket

    def is_approved_for(
        self,
        ticket_id: str,
        domain: str,
        record_id: str | int | None,
        action: str,
    ) -> bool:
        ticket = self.get_by_ticket(ticket_id)
        if ticket.status != ApprovalStatus.APPROVED:
            return False
        if ticket.domain != domain:
            return False
        if ticket.record_id and (
            record_id is None or str(ticket.record_id) != str(record_id)
        ):
            return False
        return ticket.action in {
            action,
            ApprovalActionValue.UPDATE,
            f"{ApprovalActionValue.UPDATE}:{domain}",
            ApprovalActionValue.WILDCARD,
        }