Trigger CodePipeline & Send Slack Notifications at Each Stage

import boto3
import time
import json
import urllib.request
import urllib.error
import logging
from datetime import datetime

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")


class CodePipelineMonitor:
    def __init__(
        self,
        pipeline_name: str,
        slack_webhook_url: str,
        region: str = "us-east-1",
    ):
        """
        pipeline_name:     Name of the CodePipeline to trigger and monitor.
        slack_webhook_url: Slack Incoming Webhook URL from your Slack app.
                           Format: https://hooks.slack.com/services/T.../B.../xxx
        """
        self.cp            = boto3.client("codepipeline", region_name=region)
        self.pipeline_name = pipeline_name
        self.slack_url     = slack_webhook_url
        self.region        = region

    # ── Step 1: Trigger the pipeline ─────────────────────────────
    def trigger_pipeline(self) -> str:
        """
        start_pipeline_execution() queues a new pipeline run.
        Returns a pipelineExecutionId (UUID string) that uniquely
        identifies this run — use it to query status later.

        If a run is already in progress, this starts a new parallel
        execution (CodePipeline supports concurrent executions by default).
        """
        response = self.cp.start_pipeline_execution(name=self.pipeline_name)
        execution_id = response["pipelineExecutionId"]

        logger.info(f"Pipeline triggered: {execution_id}")
        self.send_slack(
            message=(
                f"🚀 *Deployment Started*\n"
                f"Pipeline: `{self.pipeline_name}`\n"
                f"Execution ID: `{execution_id}`\n"
                f"Time: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}"
            ),
            color="#439FE0",   # Blue for "in progress"
        )
        return execution_id

    # ── Step 2: Monitor progress ──────────────────────────────────
    def monitor(
        self,
        execution_id: str,
        poll_interval: int = 15,
        timeout: int       = 1800,
    ) -> bool:
        """
        Polls the pipeline state every poll_interval seconds.
        Returns True if the pipeline succeeded, False otherwise.

        We track stage_statuses to detect changes — we only send
        a Slack notification when a stage's status actually changes,
        not on every poll cycle.

        timeout=1800 (30 min) covers most pipelines; increase for
        pipelines with slow integration tests or large deployments.
        """
        stage_statuses: dict[str, str] = {}
        start_time = time.time()

        logger.info(f"Monitoring execution {execution_id} (timeout: {timeout}s)...")

        while time.time() - start_time < timeout:
            time.sleep(poll_interval)

            try:
                # ── Check overall execution status ─────────────────
                # get_pipeline_execution() returns the high-level status:
                # InProgress / Succeeded / Failed / Stopped / Superseded
                exec_response = self.cp.get_pipeline_execution(
                    pipelineName=self.pipeline_name,
                    pipelineExecutionId=execution_id,
                )
                overall_status = exec_response["pipelineExecution"]["status"]

                # ── Check individual stage statuses ────────────────
                # get_pipeline_state() returns all stages with their
                # latestExecution status — more granular than the overall status.
                state_response = self.cp.get_pipeline_state(
                    name=self.pipeline_name
                )

                for stage in state_response["stageStates"]:
                    stage_name   = stage["stageName"]
                    latest_exec  = stage.get("latestExecution", {})
                    stage_status = latest_exec.get("status", "NotStarted")

                    # Notify only on status change (avoids spam)
                    if stage_statuses.get(stage_name) != stage_status:
                        stage_statuses[stage_name] = stage_status
                        self._notify_stage(stage_name, stage_status)

                # ── Check for terminal state ───────────────────────
                if overall_status in ("Succeeded", "Failed", "Stopped", "Superseded"):
                    self._notify_final(overall_status, execution_id)
                    return overall_status == "Succeeded"

            except self.cp.exceptions.PipelineExecutionNotFoundException:
                logger.error(f"Execution {execution_id} not found")
                return False
            except Exception as e:
                logger.error(f"Error polling pipeline: {e}")

        # Timeout reached
        self.send_slack(
            message=(
                f"⏰ *Pipeline Monitoring Timed Out*\n"
                f"Pipeline: `{self.pipeline_name}`\n"
                f"Execution: `{execution_id}`\n"
                f"Waited {timeout // 60} minutes without completion."
            ),
            color="#FFA500",  # Orange for timeout
        )
        return False

    # ── Slack notifications ───────────────────────────────────────
    def _notify_stage(self, stage_name: str, status: str) -> None:
        """Send a Slack message for each stage status change."""
        emoji_map = {
            "InProgress": "🔄",
            "Succeeded":  "✅",
            "Failed":     "❌",
            "Stopped":    "⛔",
            "Skipped":    "⏭️",
        }
        color_map = {
            "InProgress": "#439FE0",   # Blue
            "Succeeded":  "#36a64f",   # Green
            "Failed":     "#D00000",   # Red
            "Stopped":    "#FFA500",   # Orange
            "Skipped":    "#808080",   # Grey
        }
        emoji = emoji_map.get(status, "•")
        color = color_map.get(status, "#808080")

        self.send_slack(
            message=f"{emoji} Stage *{stage_name}*: `{status}`",
            color=color,
        )

    def _notify_final(self, status: str, execution_id: str) -> None:
        """Send final pipeline result notification."""
        if status == "Succeeded":
            msg   = (
                f"✅ *Deployment Succeeded!*\n"
                f"Pipeline: `{self.pipeline_name}`\n"
                f"Execution: `{execution_id}`"
            )
            color = "#36a64f"
        else:
            console_url = (
                f"https://console.aws.amazon.com/codesuite/codepipeline/pipelines/"
                f"{self.pipeline_name}/executions/{execution_id}/timeline"
            )
            msg = (
                f"❌ *Deployment FAILED!*\n"
                f"Pipeline: `{self.pipeline_name}`\n"
                f"Execution: `{execution_id}`\n"
                f"Status: `{status}`\n"
                f"<{console_url}|View in Console>"
            )
            color = "#D00000"

        self.send_slack(message=msg, color=color)

    def send_slack(self, message: str, color: str = "#439FE0") -> None:
        """
        Send a message to Slack via an Incoming Webhook.

        We use urllib.request (built-in) instead of the requests library
        to keep this script dependency-free — important for Lambda.

        Slack Incoming Webhook payload format:
        {
          "attachments": [{
            "color": "#36a64f",    ← Left-border color (hex or "good"/"warning"/"danger")
            "text": "...",         ← Message body (supports Slack mrkdwn)
            "footer": "...",       ← Small footer text
            "ts": 1234567890       ← Unix timestamp for footer display
          }]
        }

        mrkdwn formatting: *bold*, _italic_, `code`, <url|link text>
        """
        payload = {
            "attachments": [{
                "color":   color,
                "text":    message,
                "footer":  "AWS CodePipeline Monitor",
                "ts":      int(time.time()),
                "mrkdwn_in": ["text"],   # Enable mrkdwn parsing in text field
            }]
        }

        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            self.slack_url,
            data=data,
            headers={"Content-Type": "application/json"},
            method="POST",
        )

        try:
            with urllib.request.urlopen(req, timeout=5) as response:
                if response.status != 200:
                    logger.warning(f"Slack returned non-200: {response.status}")
        except urllib.error.URLError as e:
            # Log the error but don't raise — Slack failure shouldn't stop monitoring
            logger.error(f"Slack notification failed: {e}")


# ── Entry point ───────────────────────────────────────────────────
if __name__ == "__main__":
    import sys

    monitor = CodePipelineMonitor(
        pipeline_name="prod-app-pipeline",
        slack_webhook_url="https://hooks.slack.com/services/T00/B00/xxx",
        region="ap-south-1",
    )

    # Trigger and monitor
    execution_id = monitor.trigger_pipeline()
    success = monitor.monitor(
        execution_id,
        poll_interval=15,   # Check every 15 seconds
        timeout=1800,       # Give up after 30 minutes
    )

    # Exit code for CI/CD integration
    sys.exit(0 if success else 1)

self.cp            = boto3.client("codepipeline", region_name=region)
self.pipeline_name = pipeline_name
self.slack_url     = slack_webhook_url

response = self.cp.start_pipeline_execution(name=self.pipeline_name)
execution_id = response["pipelineExecutionId"]

stage_statuses: dict[str, str] = {}
start_time = time.time()

while time.time() - start_time < timeout:
    time.sleep(poll_interval)

exec_response = self.cp.get_pipeline_execution(
    pipelineName=self.pipeline_name,
    pipelineExecutionId=execution_id,
)
overall_status = exec_response["pipelineExecution"]["status"]

state_response = self.cp.get_pipeline_state(name=self.pipeline_name)
for stage in state_response["stageStates"]:
    stage_name   = stage["stageName"]
    stage_status = stage.get("latestExecution", {}).get("status", "NotStarted")

    if stage_statuses.get(stage_name) != stage_status:
        stage_statuses[stage_name] = stage_status
        self._notify_stage(stage_name, stage_status)

payload = {
    "attachments": [{
        "color":     color,
        "text":      message,
        "footer":    "AWS CodePipeline Monitor",
        "ts":        int(time.time()),
        "mrkdwn_in": ["text"],
    }]
}
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
    self.slack_url,
    data=data,
    headers={"Content-Type": "application/json"},
    method="POST",
)
with urllib.request.urlopen(req, timeout=5) as response:
    ...

success = monitor.monitor(execution_id, poll_interval=15, timeout=1800)
sys.exit(0 if success else 1)