# Modern Software Development Workflows > Implements end-to-end software development workflows including CI/CD pipelines, automated testing strategies, code quality gates, observability integration, and production deployment patterns for modern software delivery. - Skill: `paulpas/modern-software-development-workflows` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/modern-software-development-workflows` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/modern-software-development-workflows/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/modern-software-development-workflows --- # Modern Software Development Workflows Implements end-to-end software delivery workflows that connect version control through production deployment. When loaded, the model designs CI/CD pipelines with proper stage gating, configures automated testing strategies across unit/integration/e2e layers, enforces code quality gates before merge, integrates observability into every release, and applies safe deployment patterns (blue-green, canary, feature flags) to ship software with confidence and rollback capability. ## TL;DR Checklist - [ ] Verify CI pipeline has at least 4 stages: lint → test → build → deploy with explicit failure gates - [ ] Confirm branch protection rules block merges when required status checks fail - [ ] Check that unit tests run in parallel with a coverage threshold of ≥ 80% enforced as a quality gate - [ ] Validate integration tests target a disposable staging environment, not production or shared test DBs - [ ] Ensure deployment uses a safe pattern (blue-green, canary, or feature flags) with automated rollback - [ ] Verify observability metrics and health checks are injected into every release artifact before promotion - [ ] Confirm all secrets flow through a vault (e.g., SOPS, HashiCorp Vault) — never via environment variables in CI logs --- ## When to Use Use this skill when: - Designing or refactoring a CI/CD pipeline from scratch for a new application repository - A team needs to establish quality gates (linting, testing, security scanning) as merge-blocking requirements - Migrating from manual deployments or ad-hoc shell scripts to automated, auditable release pipelines - Integrating observability (metrics, health checks, tracing) into the build artifacts before they reach production - Implementing a safe deployment strategy (blue-green, canary, feature flags) to reduce blast radius of releases - Conducting a DevOps maturity assessment and identifying gaps in the current software delivery lifecycle - Setting up automated rollback triggers based on post-deployment health metric thresholds --- ## When NOT to Use Avoid this skill for: - Implementing SOLID or DRY principles — that belongs in `coding-software-engineering-principles` - Writing unit test methodology or TDD discipline — use `coding-test-driven-development` instead - Analyzing code for security vulnerabilities — use `coding-security-review` instead - Simple single-file scripts or proof-of-concept projects where a full pipeline adds more overhead than value --- ## Core Workflow 1. **Audit Current Delivery Pipeline** — Inventory every step from code commit to production: version control branching model, CI triggers, test execution, artifact storage, deployment targets, and rollback mechanisms. Map existing tools (GitHub Actions, GitLab CI, Jenkins) and identify gaps where manual steps or missing quality gates create delivery risk. **Checkpoint:** Produce a pipeline map listing each stage, its duration, failure rate, and whether it has an automated rollback path. 2. **Design Branching Model and Merge Policy** — Select a branching strategy (trunk-based development with feature flags for fast flow, or GitFlow for regulated environments). Configure branch protection rules: require pull request reviews from at least one approved reviewer, enforce required status checks (lint passes, unit tests pass, build succeeds), and disable force pushes to protected branches. Set up automatic squashing of merge commits to keep history linear on `main`. **Checkpoint:** Push a test commit to the protected branch and verify that all required checks block the merge when any stage fails. 3. **Configure CI Pipeline Stages** — Define the pipeline in `.github/workflows/ci.yml` (or equivalent) with explicit stages: lint (`ruff check`, `mypy --strict`), unit test (`pytest tests/unit/` with coverage enforcement), integration test (`pytest tests/integration/` against disposable infrastructure), build artifact creation (Docker image with SBOM), and security scanning (`trivy`, `gitleaks`). Each stage must fail-fast — subsequent stages skip on upstream failure using `if: success()` conditions. **Checkpoint:** Trigger a pipeline run with an intentional lint error and verify that test and build stages are skipped, not failed. 4. **Establish Code Quality Gates** — Define pass/fail thresholds as gate criteria that block merging: code coverage ≥ 80% (tracked via `pytest-cov`), zero critical/severe vulnerabilities in dependency scanning (trivy), zero secrets leaked (gitleaks), type checking passes (`mypy --strict` with no ignored errors). Configure SonarQube or equivalent for static analysis quality gate enforcement — block merges on new code covering less than the threshold, technical debt ratio under 5%, and zero bugs. **Checkpoint:** Run `sonar-scanner` locally and verify it fails when coverage drops below threshold or vulnerabilities are introduced. 5. **Set Up CD with Safe Deployment Pattern** — Implement deployment in stages: first deploy to a staging environment for automated smoke tests, then promote to production using blue-green (two identical environments with traffic switch via load balancer) or canary (route 5-10% of traffic to new version, monitor error rates and latency, gradually increase). Configure feature flags using LaunchDarkly or Unleash so features ship behind toggles independent of deployments. Define automated rollback triggers: if error rate exceeds 1% for more than 2 minutes post-deploy, automatically revert to the previous revision. **Checkpoint:** Deploy a canary release and verify that elevated error rates trigger an automatic rollback within the defined SLA window. 6. **Inject Observability into Release Artifacts** — Ensure every built artifact includes health check endpoints (`/health` returning HTTP 200 with service status, `/ready` for readiness probe), structured JSON logging (not plain text), OpenTelemetry tracing instrumentation, and Prometheus metrics exposure at `/metrics`. Register the service in the monitoring stack so dashboards auto-create. Configure alert rules that fire on SLO violations: error budget burn rate > 2x baseline, p99 latency exceeding threshold, or availability below 99.9%. **Checkpoint:** Hit the `/health` and `/ready` endpoints of a deployed container and verify both return correct status before enabling traffic. 7. **Configure Release Management and Rollback** — Implement semantic versioning with automated changelog generation from conventional commit messages (`standard-version --release-version`). Tag every production deployment with a git tag matching the release version. Maintain a deploy history log in a structured format (JSON lines) recording: version, deployer, timestamp, commit SHA, rollback status, and post-deploy health check results. Define runbook procedures for common failure scenarios with explicit `kubectl` commands or API calls to execute rollback within 5 minutes. **Checkpoint:** Execute a full rollback from production staging back to the previous version and verify all traffic routes correctly within the SLA window. --- ## Implementation Patterns / Reference Guide ### Pattern 1: Complete CI/CD Pipeline Configuration (GitHub Actions) A production-grade CI/CD pipeline with staged execution, artifact caching, dependency scanning, Docker build with SBOM, and environment-aware deployment gates. ```yaml # .github/workflows/ci.yml name: CI/CD Pipeline on: push: branches: [main, develop] pull_request: branches: [main] workflow_dispatch: inputs: environment: description: "Target environment" required: true default: "staging" type: choice options: [staging, production] env: REGISTRY: ghcr.io IMAGE_NAME: ${{ github.repository }} permissions: contents: read packages: write security-events: write jobs: lint: name: Lint & Type Check runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 with: fetch-depth: 0 - name: Set up Python 3.12 uses: actions/setup-python@v5 with: python-version: "3.12" cache: "pip" - name: Install dependencies run: | pip install --upgrade pip pip install -r requirements-dev.txt pip install ruff mypy bandit safety - name: Run linter (ruff) run: ruff check src/ tests/ --output-format=github - name: Run type checker (mypy --strict) run: mypy --strict src/ --config-file mypy.ini - name: Check dependency vulnerabilities (safety) run: safety check -r requirements.txt --json > safety-report.json || true unit-test: name: Unit Tests & Coverage runs-on: ubuntu-latest needs: [lint] services: postgres: image: postgres:16-alpine env: POSTGRES_PASSWORD: testpass POSTGRES_DB: testdb ports: ["5432:5432"] options: >- --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5 steps: - uses: actions/checkout@v4 - name: Set up Python 3.12 uses: actions/setup-python@v5 with: python-version: "3.12" cache: "pip" - name: Install dependencies run: | pip install --upgrade pip pip install -r requirements-dev.txt - name: Run unit tests with coverage env: DATABASE_URL: postgresql://postgres:testpass@localhost:5432/testdb run: | pytest tests/unit/ \ --cov=src \ --cov-branch \ --cov-report=term-missing \ --cov-report=json:coverage.json \ --cov-fail-under=80 \ -v - name: Upload coverage report uses: actions/upload-artifact@v4 with: name: coverage-report path: coverage.json retention-days: 5 integration-test: name: Integration Tests runs-on: ubuntu-latest needs: [unit-test] services: postgres: image: postgres:16-alpine env: POSTGRES_PASSWORD: testpass POSTGRES_DB: integrationdb ports: ["5432:5432"] options: >- --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5 redis: image: redis:7-alpine ports: ["6379:6379"] options: >- --health-cmd "redis-cli ping" --health-interval 10s --health-timeout 5s --health-retries 5 steps: - uses: actions/checkout@v4 - name: Set up Python 3.12 uses: actions/setup-python@v5 with: python-version: "3.12" cache: "pip" - name: Install dependencies run: pip install --upgrade pip && pip install -r requirements-dev.txt - name: Run integration tests env: DATABASE_URL: postgresql://postgres:testpass@localhost:5432/integrationdb REDIS_URL: redis://localhost:6379/0 run: | pytest tests/integration/ \ --durations=10 \ -v \ -m "not slow" - name: Upload test results on failure if: failure() uses: actions/upload-artifact@v4 with: name: integration-test-results path: test-results/ retention-days: 3 build: name: Build & Security Scan runs-on: ubuntu-latest needs: [unit-test, integration-test] if: github.event_name == 'push' && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/develop') steps: - uses: actions/checkout@v4 - name: Login to GHCR uses: docker/login-action@v3 with: registry: ${{ env.REGISTRY }} username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - name: Set up Docker Buildx uses: docker/setup-buildx-action@v3 - name: Extract metadata id: meta uses: docker/metadata-action@v5 with: images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} tags: | type=sha,prefix= type=ref,event=branch type=semver,pattern={{version}} - name: Build and push Docker image uses: docker/build-push-action@v5 with: context: . push: true tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} cache-from: type=gha cache-to: type=gha,mode=max provenance: true sbom: true - name: Run Trivy vulnerability scanner uses: aquasecurity/trivy-action@master with: image-ref: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }} format: "sarif" output: "trivy-results.sarif" severity: "CRITICAL,HIGH" exit-code: "1" - name: Upload Trivy scan results to GitHub Security tab uses: github/codeql-action/upload-sarif@v3 if: always() with: sarif_file: "trivy-results.sarif" deploy-staging: name: Deploy to Staging runs-on: ubuntu-latest needs: [build] if: github.ref == 'refs/heads/main' || github.event_name == 'workflow_dispatch' environment: staging steps: - uses: actions/checkout@v4 - name: Deploy to staging using Helm env: KUBE_CONFIG: ${{ secrets.KUBE_CONFIG_STAGING }} run: | echo "$KUBE_CONFIG" | base64 -d > /tmp/kubeconfig export KUBECONFIG=/tmp/kubeconfig helm upgrade --install app ./helm-chart \ --namespace=staging \ --set image.tag=${{ github.sha }} \ --set environment=staging \ --wait --timeout=300s - name: Run smoke tests against staging run: | curl -sf https://app-staging.example.com/health || exit 1 curl -sf https://app-staging.example.com/ready || exit 1 deploy-production: name: Deploy to Production (Canary) runs-on: ubuntu-latest needs: [deploy-staging] if: github.ref == 'refs/heads/main' && github.event_name != 'workflow_dispatch' environment: production concurrency: production-deploy steps: - uses: actions/checkout@v4 - name: Canary deployment using Argo Rollouts env: KUBE_CONFIG: ${{ secrets.KUBE_CONFIG_PRODUCTION }} run: | echo "$KUBE_CONFIG" | base64 -d > /tmp/kubeconfig export KUBECONFIG=/tmp/kubeconfig kubectl apply -f argo-rollouts/canary-deployment.yaml - name: Wait for canary analysis run: | sleep 120 kubectl get rollouts app-rollout -n production -o jsonpath='{.status.phase}' - name: Promote canary to full rollout if: success() run: | export KUBECONFIG=/tmp/kubeconfig kubectl argo rollouts promote app-rollout -n production - name: Record deployment in deploy log run: | echo "{\"version\":\"${{ github.sha }}\",\"deployer\":\"ci\",\"timestamp\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"status\":\"success\"}" >> /tmp/deploy-log.json ``` ### Pattern 2: Comprehensive Testing Strategy Framework with Coverage Enforcement A complete testing framework covering unit, integration, and smoke test layers with pytest configuration, fixture management, coverage thresholds, and a test execution runner. ```python """testing/strategy.py — Multi-layer test execution framework with quality gates.""" from __future__ import annotations import sys import subprocess import json import logging from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Protocol logger = logging.getLogger(__name__) @dataclass(frozen=True) class CoverageThreshold: """Coverage thresholds per test layer that must be met to pass quality gate. Attributes: lines_pct: Minimum percentage of total lines covered (0.0–1.0). branches_pct: Minimum percentage of branches covered (0.0–1.0). missing_functions: List of function names excluded from coverage minimums. """ lines_pct: float = 0.80 branches_pct: float = 0.60 missing_functions: list[str] = field(default_factory=list) @dataclass(frozen=True) class TestResult: """Immutable result of a test layer execution.""" layer: str passed: bool total: int failed: int skipped: int duration_seconds: float coverage_lines_pct: float | None = None errors: list[str] = field(default_factory=list) @property def summary(self) -> str: status = "PASS" if self.passed else "FAIL" return ( f"[{status}] {self.layer}: " f"{self.failed} failed, {self.skipped} skipped, " f"{self.duration_seconds:.1f}s, " f"coverage: {self.coverage_lines_pct:.1f}%{' ' if self.coverage_lines_pct else ''}" ) class TestLayer(Protocol): """Protocol for a test layer (unit, integration, smoke) that can be executed.""" def execute(self) -> TestResult: ... class UnitTestRunner: """Executes unit tests with coverage enforcement and parallel execution. Uses pytest-cov to measure line and branch coverage. Fails fast if the configured coverage threshold is not met. Skips integration markers by default using -m "not integration". """ def __init__( self, src_path: str = "src", test_path: str = "tests/unit", coverage_threshold: CoverageThreshold | None = None, parallel_workers: int = 4, ) -> None: self.src_path = src_path self.test_path = test_path self.coverage_threshold = coverage_threshold or CoverageThreshold() self.parallel_workers = parallel_workers def execute(self) -> TestResult: cmd = [ sys.executable, "-m", "pytest", self.test_path, f"--cov={self.src_path}", "--cov-branch", "--cov-report=json:.coverage.json", "--cov-report=term-missing", "-m", "not integration and not e2e", "-n", str(self.parallel_workers), "--dist=loadfile", "-v", ] logger.info("Running unit tests: %s", " ".join(cmd)) result = subprocess.run(cmd, capture_output=True, text=True) output_lines = result.stdout.splitlines() # Extract coverage from JSON report if available coverage_pct: float | None = None coverage_file = Path(".coverage.json") if coverage_file.exists(): with open(coverage_file) as f: cov_data = json.load(f) coverage_pct = round(cov_data.get("summary", {}).get("percent_covered", 0.0), 1) # Extract test counts from pytest output passed = sum(1 for line in output_lines if " PASSED" in line) failed = sum(1 for line in output_lines if " FAILED" in line) skipped = sum(1 for line in output_lines if " SKIPPED" in line or " xfailed" in line) duration = 0.0 for line in output_lines: if "Total test time:" in line or "duration:" in line.lower(): parts = line.split(":") if len(parts) >= 2: try: duration = float(parts[-1].strip().split(" ")[0]) except ValueError: pass passed = result.returncode == 0 errors = [l.strip() for l in result.stderr.splitlines()[-10:] if l.strip()] if not passed else [] return TestResult( layer="unit", passed=passed and (coverage_pct is None or coverage_pct >= self.coverage_threshold.lines_pct), total=passed + failed + skipped, failed=failed if not passed else 0, skipped=skipped, duration_seconds=duration, coverage_lines_pct=coverage_pct, errors=errors, ) class IntegrationTestRunner: """Executes integration tests against disposable service instances. Spins up required services via Docker Compose, runs integration test suite, then tears down all infrastructure. Each test is isolated with unique database schemas or namespaces to prevent cross-test contamination. """ def __init__(self, compose_file: str = "docker-compose.test.yml") -> None: self.compose_file = compose_file self.services_dir = Path("/tmp/test-services") def execute(self) -> TestResult: # Start disposable infrastructure subprocess.run( ["docker", "compose", "-f", self.compose_file, "up", "-d"], check=True, capture_output=True, text=True, ) try: cmd = [ sys.executable, "-m", "pytest", "tests/integration/", "-v", "-m", "integration", "--timeout=60", ] result = subprocess.run(cmd, capture_output=True, text=True) passed = result.returncode == 0 output_lines = result.stdout.splitlines() if result.stdout else [] failed = sum(1 for l in output_lines if " FAILED" in l) skipped = sum(1 for l in output_lines if " SKIPPED" in l or " xfailed" in l) return TestResult( layer="integration", passed=passed, total=len(output_lines), failed=failed if not passed else 0, skipped=skipped, duration_seconds=0.0, errors=[l.strip() for l in result.stderr.splitlines()[-5:]] if not passed else [], ) finally: subprocess.run( ["docker", "compose", "-f", self.compose_file, "down", "--volumes"], check=False, capture_output=True, text=True, ) def run_pipeline(layers: list[TestLayer], fail_fast: bool = True) -> list[TestResult]: """Execute test layers in sequence with optional fail-fast behavior. Returns a list of TestResults for each layer. If fail_fast is True, stops executing remaining layers when a prior layer fails. Args: layers: Ordered list of test layer runners to execute. fail_fast: When True, skip subsequent layers if any layer fails. Returns: List of TestResult objects in execution order. """ results: list[TestResult] = [] for layer_runner in layers: result = layer_runner.execute() results.append(result) logger.info("Layer complete: %s", result.summary) if not result.passed and fail_fast: logger.warning("Fail-fast triggered — skipping remaining layers after %s failure", result.layer) break return results # --- BAD vs GOOD: Coverage Enforcement Pattern --- # ❌ BAD: Coverage check is a separate manual step with no enforcement. # The team runs coverage locally but CI allows merging regardless of threshold. def bad_coverage_check() -> None: # type: ignore[no-untyped-def] import subprocess subprocess.run(["pytest", "--cov=src", "tests/unit/"]) print("Coverage run complete — please verify it's above 80%") # No enforcement! # ✅ GOOD: Coverage threshold is enforced programmatically with clear failure mode. # pytest-cov's --cov-fail-under exits non-zero when below the threshold, # causing CI to block the merge automatically. def good_coverage_check(threshold: float = 0.80) -> None: """Run unit tests and enforce minimum coverage threshold. Args: threshold: Minimum required coverage percentage (e.g., 0.80 for 80%). Raises: RuntimeError: If coverage falls below the enforced threshold. """ import subprocess result = subprocess.run( [sys.executable, "-m", "pytest", "tests/unit/", f"--cov=src", "--cov-branch", f"--cov-fail-under={int(threshold * 100)}", "-v"], capture_output=True, text=True, ) if result.returncode != 0: coverage_line = [l for l in result.stdout.splitlines() if "TOTAL" in l] actual = coverage_line[0].split()[-1] if coverage_line else "unknown" raise RuntimeError( f"Coverage gate FAILED: {actual} is below the {int(threshold * 100)}% threshold.\n" f"Add tests or update src/ to improve coverage." ) print(f"Coverage gate PASSED: >= {int(threshold * 100)}%") ``` ### Pattern 3: Code Quality Gate Enforcement with Multi-Tool Scanning Enforces quality gates across multiple tools (ruff, mypy, bandit, safety, trivy) with a unified pass/fail report and configurable severity thresholds. ```python """quality/gates.py — Unified code quality gate enforcement for CI merge protection.""" from __future__ import annotations import json import logging import subprocess import sys from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Protocol logger = logging.getLogger(__name__) class Severity(str, Enum): """Severity levels for quality gate findings.""" CRITICAL = "critical" HIGH = "high" MEDIUM = "medium" LOW = "low" @dataclass(frozen=True) class GateConfig: """Configuration for a single quality gate. Attributes: tool: Name of the scanning tool (ruff, mypy, bandit, safety, trivy). severity_threshold: Minimum severity to consider as a violation. allow_fixme: If True, findings marked with "noqa" or "# type: ignore" are exempt. """ tool: str severity_threshold: Severity = Severity.HIGH allow_fixme: bool = True @dataclass(frozen=True) class GateResult: """Result from a single quality gate evaluation.""" config: GateConfig passed: bool findings_count: int critical_count: int high_count: int details: list[str] = field(default_factory=list) def __str__(self) -> str: status = "PASS" if self.passed else "FAIL" return ( f"[{status}] {self.config.tool}: " f"{self.findings_count} findings " f"({self.critical_count} critical, {self.high_count} high)" ) class QualityTool(Protocol): """Protocol for a quality scanning tool.""" def scan(self, config: GateConfig) -> GateResult: ... class RuffScanner: """Runs ruff linter and checks for violations above the severity threshold. Uses ruff's --output-format=json output to parse individual findings with their severity levels (E = error/warning, F = fatal). """ def __init__(self, src_dirs: list[str] | None = None) -> None: self.src_dirs = src_dirs or ["src/", "tests/"] def scan(self, config: GateConfig) -> GateResult: if config.tool != "ruff": return GateResult( config=config, passed=True, findings_count=0, critical_count=0, high_count=0, details=["Scanning skipped: tool mismatch"], ) cmd = [sys.executable, "-m", "ruff", "check"] + self.src_dirs + [ "--output-format=json", ] result = subprocess.run(cmd, capture_output=True, text=True) findings: list[dict] = [] if result.stdout.strip(): try: findings = json.loads(result.stdout) except json.JSONDecodeError: return GateResult( config=config, passed=False, findings_count=1, critical_count=1, high_count=0, details=[f"Failed to parse ruff JSON output: {result.stderr[:200]}"], ) severity_map = {"F": Severity.CRITICAL, "E": Severity.HIGH, "W": Severity.MEDIUM} threshold_order = {Severity.CRITICAL: 0, Severity.HIGH: 1, Severity.MEDIUM: 2, Severity.LOW: 3} max_severity_idx = threshold_order.get(config.severity_threshold, 1) critical_count = 0 high_count = 0 details: list[str] = [] for finding in findings: code = finding.get("code", "") rule_type = code[0] if code else "" severity = severity_map.get(rule_type, Severity.LOW) idx = threshold_order.get(severity, 3) if config.allow_fixme and "noqa" in finding.get("message", "").lower(): continue details.append(f"{finding['filename']}:{finding.get('line_no', '?')}:{finding.get('column', '?')} [{code}] {finding.get('message', '')}") if idx <= max_severity_idx: if severity == Severity.CRITICAL: critical_count += 1 elif severity == Severity.HIGH: high_count += 1 total_violations = sum( 1 for f in findings if not (config.allow_fixme and "noqa" in f.get("message", "").lower()) ) threshold_foundings = critical_count + high_count passed = threshold_foundings == 0 return GateResult( config=config, passed=passed, findings_count=total_violations, critical_count=critical_count, high_count=high_count, details=details[:10], # Cap details to avoid flooding output ) class MypyScanner: """Runs mypy type checker with strict mode and checks for violations.""" def __init__(self, src_dirs: list[str] | None = None) -> None: self.src_dirs = src_dirs or ["src/"] def scan(self, config: GateConfig) -> GateResult: if config.tool != "mypy": return GateResult( config=config, passed=True, findings_count=0, critical_count=0, high_count=0, details=["Scanning skipped: tool mismatch"], ) cmd = [sys.executable, "-m", "mypy"] + self.src_dirs + [ "--strict", "--show-error-codes", "--show-traceback", ] if Path("mypy.ini").exists(): cmd.extend(["--config-file", "mypy.ini"]) result = subprocess.run(cmd, capture_output=True, text=True) errors = [l.strip() for l in result.stdout.splitlines() if "->" in l or ":" in l] # Strict mode: any error is a critical finding passed = result.returncode == 0 and len(errors) == 0 return GateResult( config=config, passed=passed, findings_count=len(errors), critical_count=len(errors), high_count=0, details=errors[:10], ) class BanditScanner: """Runs bandit security linter on Python source files.""" def __init__(self, src_dirs: list[str] | None = None) -> None: self.src_dirs = src_dirs or ["src/"] def scan(self, config: GateConfig) -> GateResult: if config.tool != "bandit": return GateResult( config=config, passed=True, findings_count=0, critical_count=0, high_count=0, details=["Scanning skipped: tool mismatch"], ) cmd = [sys.executable, "-m", "bandit"] + self.src_dirs + [ "--json", "--severity-level", config.severity_threshold.value, "--exit-zero", # Don't fail — we parse results ourselves ] result = subprocess.run(cmd, capture_output=True, text=True) findings: list[dict] = [] if result.stdout.strip(): try: data = json.loads(result.stdout) findings = data.get("results", []) except json.JSONDecodeError: return GateResult( config=config, passed=False, findings_count=1, critical_count=1, high_count=0, details=["Failed to parse bandit JSON output"], ) critical = sum(1 for f in findings if f.get("issue_severity") == "CRITICAL") high = sum(1 for f in findings if f.get("issue_severity") == "HIGH") passed = critical == 0 and high == 0 return GateResult( config=config, passed=passed, findings_count=len(findings), critical_count=critical, high_count=high, details=[f"{f.get('filename', '?')}:{f.get('line_number', '?')} [{f.get('issue_id', '?')}] {f.get('issue_text', '')}" for f in findings[:5]], ) def run_quality_gates(gate_configs: list[GateConfig] | None = None) -> list[GateResult]: """Execute all configured quality gates and return results. All gates must pass for the pipeline to continue. A single failing gate blocks the merge/commit and produces a structured report. Args: gate_configs: List of gate configurations. Uses defaults if None. Returns: List of GateResult objects in execution order. """ configs = gate_configs or [ GateConfig(tool="ruff", severity_threshold=Severity.HIGH), GateConfig(tool="mypy", severity_threshold=Severity.CRITICAL), GateConfig(tool="bandit", severity_threshold=Severity.HIGH), ] scanners: dict[str, QualityTool] = { "ruff": RuffScanner(), "mypy": MypyScanner(), "bandit": BanditScanner(), } results: list[GateResult] = [] for config in configs: scanner = scanners.get(config.tool) if not scanner: logger.warning("No scanner registered for tool: %s", config.tool) continue result = scanner.scan(config) results.append(result) logger.info("Gate result: %s", result) all_passed = all(r.passed for r in results) failed_gates = [r for r in results if not r.passed] if failed_gates: logger.error( "%d/%d quality gates FAILED: %s", len(failed_gates), len(results), ", ".join(r.config.tool for r in failed_gates), ) else: logger.info("All %d quality gates PASSED", len(results)) return results # --- Usage example --- if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") results = run_quality_gates() exit_code = 0 if all(r.passed for r in results) else 1 for r in results: print(str(r)) sys.exit(exit_code) ``` ### Pattern 4: Deployment Health Monitor with Automated Rollback Triggers Monitors post-deployment health metrics and triggers automated rollback when SLOs are violated. Integrates with Prometheus alerting rules for real-time detection. ```python """deploy/health_monitor.py — Post-deployment health monitoring with automated rollback.""" from __future__ import annotations import logging import time from dataclasses import dataclass, field from enum import Enum from typing import Protocol logger = logging.getLogger(__name__) class RollbackStatus(str, Enum): """Status of a rollback operation.""" PENDING = "pending" IN_PROGRESS = "in_progress" COMPLETED = "completed" FAILED = "failed" NOT_NEEDED = "not_needed" @dataclass(frozen=True) class SLOViolation: """Recorded SLO violation that triggered rollback consideration. Attributes: metric: The Prometheus metric name that violated its threshold. current_value: The observed value at violation time. threshold: The configured alerting threshold. duration_seconds: How long the violation persisted before triggering. window_minutes: The evaluation window for the SLO check. """ metric: str current_value: float threshold: float duration_seconds: int window_minutes: int @dataclass(frozen=True) class HealthCheckResult: """Result of a single health check against a deployed service. Attributes: endpoint: The URL or path checked. http_status: HTTP status code returned (0 if unreachable). response_time_ms: Time taken for the request in milliseconds. passed: True when all health criteria are met. """ endpoint: str http_status: int response_time_ms: float passed: bool class HealthChecker(Protocol): """Protocol for checking service health endpoints.""" def check_health(self, service_name: str) -> list[HealthCheckResult]: ... def check_slo_metrics(self, service_name: str, window_minutes: int = 5) -> list[SLOViolation]: ... class PrometheusHealthChecker: """Checks service health by querying Prometheus for metrics and HTTP endpoints. Uses the Prometheus API to evaluate SLO conditions: error budget burn rate, p99 latency, and availability percentage. Also performs HTTP-level health check probing against the deployed /health endpoint. """ def __init__(self, prometheus_url: str, timeout_seconds: float = 10.0) -> None: self.prometheus_url = prometheus_url.rstrip("/") self.timeout_seconds = timeout_seconds def _query_prometheus(self, query: str) -> dict | list[dict]: # type: ignore[type-arg] """Execute a PromQL query against the Prometheus API.""" import urllib.request import json url = f"{self.prometheus_url}/api/v1/query?query={urllib.parse.quote(query)}" req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=self.timeout_seconds) as resp: data = json.loads(resp.read()) if data.get("status") != "success": raise RuntimeError(f"Prometheus query failed: {data}") return data["data"] def check_health(self, service_name: str) -> list[HealthCheckResult]: """Check HTTP health and readiness endpoints for the service. Args: service_name: The Kubernetes service name to probe. Returns: List of health check results for each endpoint checked. """ import urllib.request import urllib.error endpoints = ["/health", "/ready"] base_url = f"http://{service_name}" results: list[HealthCheckResult] = [] for endpoint in endpoints: url = f"{base_url}{endpoint}" start = time.monotonic() try: req = urllib.request.Request(url, method="GET") with urllib.request.urlopen(req, timeout=self.timeout_seconds) as resp: status = resp.status elapsed_ms = (time.monotonic() - start) * 1000 results.append(HealthCheckResult( endpoint=url, http_status=status, response_time_ms=round(elapsed_ms, 2), passed=status == 200 and elapsed_ms < 500, )) except (urllib.error.URLError, TimeoutError) as e: results.append(HealthCheckResult( endpoint=url, http_status=0, response_time_ms=round((time.monotonic() - start) * 1000, 2), passed=False, )) logger.warning("Health check failed for %s: %s", url, e) return results def check_slo_metrics(self, service_name: str, window_minutes: int = 5) -> list[SLOViolation]: """Check SLO metrics via PromQL queries. Monitors error budget burn rate, p99 latency, and availability. Triggers on sustained violations for the evaluation window. Args: service_name: Service identifier matching Prometheus labels. window_minutes: Time window for SLO evaluation in minutes. Returns: List of SLO violations found (empty if all SLOs are met). """ now = int(time.time()) queries = { "error_rate": ( f'sum(rate(http_requests_total{{service="{service_name}",' f'code=~"5.."}}[{window_minutes}m])) / ' f'sum(rate(http_requests_total{{service="{service_name}"}}[{windo …(truncated)