Implementing CI/CD with Python

# Our CI/CD Pipeline Flow: # 1. Code Push → Git Repository # 2. Trigger → CI Pipeline # 3. Test → Unit Tests, Integration Tests, UAT # 4. Build → Docker Container # 5. Deploy → Staging Environment # 6. Validate → Sanity Checks # 7. Deploy → Production (Blue-Green) # 8. Monitor → Health Checks # 9. Rollback → If Issues Detected

python-cicd-demo/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── models/ │ ├── routes/ │ ├── services/ │ └── utils/ ├── tests/ │ ├── __init__.py │ ├── unit/ │ ├── integration/ │ └── uat/ ├── docker/ │ ├── Dockerfile │ ├── docker-compose.yml │ └── nginx.conf ├── scripts/ │ ├── deploy.sh │ ├── rollback.sh │ └── health_check.py ├── .github/ │ └── workflows/ │ ├── ci.yml │ └── cd.yml ├── requirements.txt ├── requirements-dev.txt ├── pytest.ini ├── setup.cfg ├── .gitignore └── README.md

# requirements.txt flask==2.3.3 gunicorn==21.2.0 psycopg2-binary==2.9.7 redis==4.6.0 requests==2.31.0 python-dotenv==1.0.0 prometheus-client==0.17.1 # requirements-dev.txt pytest==7.4.2 pytest-cov==4.1.0 pytest-mock==3.11.1 black==23.7.0 flake8==6.0.0 mypy==1.5.1 bandit==1.7.5 safety==2.3.4 pre-commit==3.4.0

# .gitignore __pycache__/ *.py[cod] *$py.class *.so .Python build/ develop-eggs/ dist/ downloads/ eggs/ .eggs/ lib/ lib64/ parts/ sdist/ var/ wheels/ *.egg-info/ .installed.cfg *.egg # Virtual environments venv/ env/ ENV/ # IDE .vscode/ .idea/ *.swp *.swo # Environment variables .env .env.local .env.production # Testing .coverage htmlcov/ .pytest_cache/ .tox/ # Docker .dockerignore # Logs *.log logs/

# app/main.py from flask import Flask, jsonify, request from prometheus_client import Counter, Histogram, generate_latest import os import time import logging # Metrics REQUEST_COUNT = Counter('app_requests_total', 'Total requests', ['method', 'endpoint']) REQUEST_LATENCY = Histogram('app_request_duration_seconds', 'Request latency') def create_app(): app = Flask(__name__) # Configuration app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'dev-secret-key') app.config['DATABASE_URL'] = os.getenv('DATABASE_URL', 'sqlite:///app.db') # Logging logging.basicConfig(level=logging.INFO) @app.before_request def before_request(): request.start_time = time.time() @app.after_request def after_request(response): REQUEST_COUNT.labels(method=request.method, endpoint=request.endpoint).inc() REQUEST_LATENCY.observe(time.time() - request.start_time) return response # Routes @app.route('/') def home(): return jsonify({ 'message': 'Python CI/CD Demo API', 'version': os.getenv('APP_VERSION', '1.0.0'), 'status': 'healthy' }) @app.route('/health') def health_check(): return jsonify({ 'status': 'healthy', 'timestamp': time.time(), 'version': os.getenv('APP_VERSION', '1.0.0') }) @app.route('/metrics') def metrics(): return generate_latest() @app.route('/api/users', methods=['GET', 'POST']) def users(): if request.method == 'GET': return jsonify({'users': []}) elif request.method == 'POST': user_data = request.get_json() return jsonify({'message': 'User created', 'user': user_data}), 201 return app if __name__ == '__main__': app = create_app() app.run(host='0.0.0.0', port=int(os.getenv('PORT', 5000)), debug=False)

# app/services/user_service.py import logging from typing import List, Dict, Optional logger = logging.getLogger(__name__) class UserService: def __init__(self): self.users = [] # In production, this would be a database def create_user(self, user_data: Dict) -> Dict: """Create a new user""" try: user = { 'id': len(self.users) + 1, 'name': user_data.get('name'), 'email': user_data.get('email'), 'created_at': time.time() } self.users.append(user) logger.info(f"User created: {user['id']}") return user except Exception as e: logger.error(f"Error creating user: {str(e)}") raise def get_users(self) -> List[Dict]: """Get all users""" return self.users def get_user_by_id(self, user_id: int) -> Optional[Dict]: """Get user by ID""" return next((user for user in self.users if user['id'] == user_id), None) def validate_user_data(self, user_data: Dict) -> bool: """Validate user data""" required_fields = ['name', 'email'] return all(field in user_data and user_data[field] for field in required_fields)

# pytest.ini [tool:pytest] testpaths = tests python_files = test_*.py python_classes = Test* python_functions = test_* addopts = --verbose --cov=app --cov-report=html --cov-report=xml --cov-report=term-missing --cov-fail-under=80 --junitxml=test-results.xml markers = unit: Unit tests integration: Integration tests uat: User acceptance tests slow: Slow running tests

# tests/unit/test_user_service.py import pytest from app.services.user_service import UserService class TestUserService: @pytest.fixture def user_service(self): return UserService() @pytest.fixture def valid_user_data(self): return { 'name': 'John Doe', 'email': 'john@example.com' } def test_create_user_success(self, user_service, valid_user_data): """Test successful user creation""" user = user_service.create_user(valid_user_data) assert user['id'] == 1 assert user['name'] == 'John Doe' assert user['email'] == 'john@example.com' assert 'created_at' in user def test_create_user_increments_id(self, user_service, valid_user_data): """Test that user IDs increment correctly""" user1 = user_service.create_user(valid_user_data) user2 = user_service.create_user(valid_user_data) assert user1['id'] == 1 assert user2['id'] == 2 def test_get_users_empty_initially(self, user_service): """Test that users list is empty initially""" users = user_service.get_users() assert users == [] def test_get_users_returns_created_users(self, user_service, valid_user_data): """Test that get_users returns created users""" user_service.create_user(valid_user_data) users = user_service.get_users() assert len(users) == 1 assert users[0]['name'] == 'John Doe' def test_get_user_by_id_existing(self, user_service, valid_user_data): """Test getting existing user by ID""" created_user = user_service.create_user(valid_user_data) found_user = user_service.get_user_by_id(created_user['id']) assert found_user == created_user def test_get_user_by_id_nonexistent(self, user_service): """Test getting non-existent user by ID""" user = user_service.get_user_by_id(999) assert user is None def test_validate_user_data_valid(self, user_service, valid_user_data): """Test validation with valid user data""" assert user_service.validate_user_data(valid_user_data) is True @pytest.mark.parametrize("invalid_data", [ {}, {'name': 'John'}, {'email': 'john@example.com'}, {'name': '', 'email': 'john@example.com'}, {'name': 'John', 'email': ''} ]) def test_validate_user_data_invalid(self, user_service, invalid_data): """Test validation with invalid user data""" assert user_service.validate_user_data(invalid_data) is False

# tests/unit/test_main.py import pytest import json from app.main import create_app class TestFlaskApp: @pytest.fixture def app(self): app = create_app() app.config['TESTING'] = True return app @pytest.fixture def client(self, app): return app.test_client() def test_home_endpoint(self, client): """Test home endpoint returns correct response""" response = client.get('/') assert response.status_code == 200 data = json.loads(response.data) assert data['message'] == 'Python CI/CD Demo API' assert data['status'] == 'healthy' assert 'version' in data def test_health_check_endpoint(self, client): """Test health check endpoint""" response = client.get('/health') assert response.status_code == 200 data = json.loads(response.data) assert data['status'] == 'healthy' assert 'timestamp' in data assert 'version' in data def test_metrics_endpoint(self, client): """Test metrics endpoint returns prometheus metrics""" response = client.get('/metrics') assert response.status_code == 200 assert b'app_requests_total' in response.data def test_users_get_endpoint(self, client): """Test GET /api/users endpoint""" response = client.get('/api/users') assert response.status_code == 200 data = json.loads(response.data) assert 'users' in data assert isinstance(data['users'], list) def test_users_post_endpoint(self, client): """Test POST /api/users endpoint""" user_data = {'name': 'Test User', 'email': 'test@example.com'} response = client.post('/api/users', data=json.dumps(user_data), content_type='application/json') assert response.status_code == 201 data = json.loads(response.data) assert data['message'] == 'User created' assert data['user'] == user_data

# tests/integration/test_database_integration.py import pytest import os import tempfile from app.main import create_app class TestDatabaseIntegration: @pytest.fixture def app(self): # Create temporary database for testing db_fd, db_path = tempfile.mkstemp() app = create_app() app.config['TESTING'] = True app.config['DATABASE_URL'] = f'sqlite:///{db_path}' yield app # Cleanup os.close(db_fd) os.unlink(db_path) @pytest.fixture def client(self, app): return app.test_client() @pytest.mark.integration def test_user_creation_flow(self, client): """Test complete user creation flow with database""" # Create user user_data = {'name': 'Integration Test User', 'email': 'integration@test.com'} response = client.post('/api/users', json=user_data) assert response.status_code == 201 # Verify user exists response = client.get('/api/users') assert response.status_code == 200 # Additional database verification would go here

# tests/integration/test_api_integration.py import pytest import requests import time from concurrent.futures import ThreadPoolExecutor class TestAPIIntegration: @pytest.fixture def base_url(self): return os.getenv('TEST_API_URL', 'http://localhost:5000') @pytest.mark.integration def test_api_health_check(self, base_url): """Test API health check endpoint""" response = requests.get(f'{base_url}/health') assert response.status_code == 200 data = response.json() assert data['status'] == 'healthy' assert 'timestamp' in data @pytest.mark.integration def test_api_load_handling(self, base_url): """Test API can handle concurrent requests""" def make_request(): return requests.get(f'{base_url}/health') # Make 10 concurrent requests with ThreadPoolExecutor(max_workers=10) as executor: futures = [executor.submit(make_request) for _ in range(10)] responses = [future.result() for future in futures] # All requests should succeed assert all(r.status_code == 200 for r in responses) @pytest.mark.integration @pytest.mark.slow def test_api_response_time(self, base_url): """Test API response time is acceptable""" start_time = time.time() response = requests.get(f'{base_url}/health') end_time = time.time() assert response.status_code == 200 assert (end_time - start_time) < 1.0 # Should respond within 1 second

# tests/uat/test_user_acceptance.py import pytest import requests import json from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC class TestUserAcceptance: @pytest.fixture def api_url(self): return os.getenv('UAT_API_URL', 'http://localhost:5000') @pytest.fixture def web_driver(self): options = webdriver.ChromeOptions() options.add_argument('--headless') options.add_argument('--no-sandbox') options.add_argument('--disable-dev-shm-usage') driver = webdriver.Chrome(options=options) yield driver driver.quit() @pytest.mark.uat def test_user_can_access_api(self, api_url): """UAT: User can access the API and get a response""" response = requests.get(f'{api_url}/') assert response.status_code == 200 data = response.json() assert 'message' in data assert 'status' in data assert data['status'] == 'healthy' @pytest.mark.uat def test_user_can_create_and_retrieve_users(self, api_url): """UAT: User can create and retrieve users via API""" # Create a user user_data = { 'name': 'UAT Test User', 'email': 'uat@example.com' } create_response = requests.post( f'{api_url}/api/users', json=user_data, headers={'Content-Type': 'application/json'} ) assert create_response.status_code == 201 create_data = create_response.json() assert create_data['message'] == 'User created' assert create_data['user']['name'] == user_data['name'] # Retrieve users get_response = requests.get(f'{api_url}/api/users') assert get_response.status_code == 200 get_data = get_response.json() assert 'users' in get_data @pytest.mark.uat def test_api_error_handling(self, api_url): """UAT: API handles errors gracefully""" # Test invalid endpoint response = requests.get(f'{api_url}/invalid-endpoint') assert response.status_code == 404 # Test invalid JSON response = requests.post( f'{api_url}/api/users', data='invalid json', headers={'Content-Type': 'application/json'} ) assert response.status_code in [400, 422] @pytest.mark.uat def test_system_performance_requirements(self, api_url): """UAT: System meets performance requirements""" import time # Test response time requirement start_time = time.time() response = requests.get(f'{api_url}/health') end_time = time.time() assert response.status_code == 200 assert (end_time - start_time) < 2.0 # UAT requirement: < 2 seconds # Test multiple requests for _ in range(5): response = requests.get(f'{api_url}/health') assert response.status_code == 200

# Dockerfile FROM python:3.11-slim as builder # Set environment variables ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ PIP_NO_CACHE_DIR=1 \ PIP_DISABLE_PIP_VERSION_CHECK=1 # Install system dependencies RUN apt-get update && apt-get install -y \ build-essential \ libpq-dev \ && rm -rf /var/lib/apt/lists/* # Create virtual environment RUN python -m venv /opt/venv ENV PATH="/opt/venv/bin:$PATH" # Copy requirements and install Python dependencies COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Production stage FROM python:3.11-slim as production # Set environment variables ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ PATH="/opt/venv/bin:$PATH" # Install runtime dependencies RUN apt-get update && apt-get install -y \ libpq5 \ curl \ && rm -rf /var/lib/apt/lists/* # Copy virtual environment from builder stage COPY --from=builder /opt/venv /opt/venv # Create non-root user RUN groupadd -r appuser && useradd -r -g appuser appuser # Set work directory WORKDIR /app # Copy application code COPY app/ ./app/ COPY scripts/ ./scripts/ # Change ownership to non-root user RUN chown -R appuser:appuser /app USER appuser # Health check HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \ CMD curl -f http://localhost:5000/health || exit 1 # Expose port EXPOSE 5000 # Run application CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "--timeout", "120", "app.main:create_app()"]

# scripts/health_check.py import requests import time import sys import logging from typing import Dict, List, Tuple logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class HealthChecker: def __init__(self, base_url: str, timeout: int = 30): self.base_url = base_url.rstrip('/') self.timeout = timeout self.checks = [] def add_check(self, name: str, endpoint: str, expected_status: int = 200, expected_content: str = None, max_response_time: float = 5.0): """Add a health check""" self.checks.append({ 'name': name, 'endpoint': endpoint, 'expected_status': expected_status, 'expected_content': expected_content, 'max_response_time': max_response_time }) def run_check(self, check: Dict) -> Tuple[bool, str]: """Run a single health check""" try: start_time = time.time() response = requests.get( f"{self.base_url}{check['endpoint']}", timeout=self.timeout ) response_time = time.time() - start_time # Check status code if response.status_code != check['expected_status']: return False, f"Expected status {check['expected_status']}, got {response.status_code}" # Check response time if response_time > check['max_response_time']: return False, f"Response time {response_time:.2f}s exceeds limit {check['max_response_time']}s" # Check content if specified if check['expected_content']: if check['expected_content'] not in response.text: return False, f"Expected content '{check['expected_content']}' not found" return True, f"OK (response time: {response_time:.2f}s)" except requests.exceptions.RequestException as e: return False, f"Request failed: {str(e)}" except Exception as e: return False, f"Unexpected error: {str(e)}" def run_all_checks(self) -> bool: """Run all health checks""" logger.info(f"🏥 Running health checks against {self.base_url}") all_passed = True results = [] for check in self.checks: logger.info(f"🔍 Checking {check['name']}...") passed, message = self.run_check(check) if passed: logger.info(f"✅ {check['name']}: {message}") else: logger.error(f"❌ {check['name']}: {message}") all_passed = False results.append({ 'name': check['name'], 'passed': passed, 'message': message }) # Summary passed_count = sum(1 for r in results if r['passed']) total_count = len(results) if all_passed: logger.info(f"🎉 All {total_count} health checks passed!") else: logger.error(f"💥 {total_count - passed_count} of {total_count} health checks failed!") return all_passed def main(): import os # Configuration base_url = os.getenv('HEALTH_CHECK_URL', 'http://localhost:5000') # Initialize health checker checker = HealthChecker(base_url) # Add health checks checker.add_check('Basic Health', '/health', expected_content='healthy') checker.add_check('API Root', '/', expected_content='Python CI/CD Demo API') checker.add_check('Users API', '/api/users', expected_status=200) checker.add_check('Metrics', '/metrics', expected_content='app_requests_total') # Run checks success = checker.run_all_checks() # Exit with appropriate code sys.exit(0 if success else 1) if __name__ == '__main__': main()

# scripts/monitoring.py import requests import time import json import os from prometheus_client.parser import text_string_to_metric_families class ApplicationMonitor: def __init__(self, app_url: str, alert_webhook: str = None): self.app_url = app_url.rstrip('/') self.alert_webhook = alert_webhook self.thresholds = { 'response_time': 2.0, # seconds 'error_rate': 0.05, # 5% 'memory_usage': 0.8, # 80% 'cpu_usage': 0.8 # 80% } def check_response_time(self) -> Dict: """Check application response time""" try: start_time = time.time() response = requests.get(f'{self.app_url}/health', timeout=10) response_time = time.time() - start_time is_healthy = ( response.status_code == 200 and response_time < self.thresholds['response_time'] ) return { 'metric': 'response_time', 'value': response_time, 'threshold': self.thresholds['response_time'], 'healthy': is_healthy, 'message': f"Response time: {response_time:.2f}s" } except Exception as e: return { 'metric': 'response_time', 'value': None, 'threshold': self.thresholds['response_time'], 'healthy': False, 'message': f"Health check failed: {str(e)}" } def check_error_rate(self) -> Dict: """Check application error rate from metrics""" try: response = requests.get(f'{self.app_url}/metrics', timeout=10) if response.status_code != 200: raise Exception(f"Metrics endpoint returned {response.status_code}") # Parse Prometheus metrics total_requests = 0 error_requests = 0 for family in text_string_to_metric_families(response.text): if family.name == 'app_requests_total': for sample in family.samples: total_requests += sample.value # Count 4xx and 5xx as errors if sample.labels.get('status', '').startswith(('4', '5')): error_requests += sample.value error_rate = error_requests / total_requests if total_requests > 0 else 0 is_healthy = error_rate < self.thresholds['error_rate'] return { 'metric': 'error_rate', 'value': error_rate, 'threshold': self.thresholds['error_rate'], 'healthy': is_healthy, 'message': f"Error rate: {error_rate:.2%} ({error_requests}/{total_requests})" } except Exception as e: return { 'metric': 'error_rate', 'value': None, 'threshold': self.thresholds['error_rate'], 'healthy': False, 'message': f"Error rate check failed: {str(e)}" } def send_alert(self, alert_data: Dict): """Send alert to webhook""" if not self.alert_webhook: return try: payload = { 'text': f"🚨 Alert: {alert_data['message']}", 'attachments': [{ 'color': 'danger', 'fields': [ {'title': 'Metric', 'value': alert_data['metric'], 'short': True}, {'title': 'Value', 'value': str(alert_data['value']), 'short': True}, {'title': 'Threshold', 'value': str(alert_data['threshold']), 'short': True}, {'title': 'Timestamp', 'value': time.strftime('%Y-%m-%d %H:%M:%S'), 'short': True} ] }] } requests.post(self.alert_webhook, json=payload, timeout=10) except Exception as e: print(f"Failed to send alert: {str(e)}") def run_monitoring_cycle(self): """Run a complete monitoring cycle""" checks = [ self.check_response_time(), self.check_error_rate() ] for check in checks: if not check['healthy']: print(f"❌ {check['message']}") self.send_alert(check) else: print(f"✅ {check['message']}") return all(check['healthy'] for check in checks) def main(): app_url = os.getenv('APP_URL', 'http://localhost:5000') alert_webhook = os.getenv('ALERT_WEBHOOK_URL') monitor = ApplicationMonitor(app_url, alert_webhook) # Run monitoring is_healthy = monitor.run_monitoring_cycle() if not is_healthy: print("🚨 Application health issues detected!") exit(1) else: print("✅ Application is healthy!") if __name__ == '__main__': main()

# scripts/blue_green_deploy.py import subprocess import time import requests import sys import os class BlueGreenDeployment: def __init__(self, app_name: str, new_version: str): self.app_name = app_name self.new_version = new_version self.current_color = self.get_current_color() self.target_color = 'green' if self.current_color == 'blue' else 'blue' def get_current_color(self) -> str: """Determine current active color""" try: # Check which color is currently receiving traffic result = subprocess.run(['docker', 'ps', '--format', 'table {{.Names}}'], capture_output=True, text=True) if f'{self.app_name}-blue' in result.stdout: return 'blue' elif f'{self.app_name}-green' in result.stdout: return 'green' else: return 'blue' # Default to blue if neither is running except: return 'blue' def deploy_target_environment(self): """Deploy new version to target environment""" print(f"🚀 Deploying {self.new_version} to {self.target_color} environment") # Stop target environment if running subprocess.run([ 'docker-compose', '-f', f'docker-compose.{self.target_color}.yml', 'down' ], check=False) # Start target environment with new version env = os.environ.copy() env['APP_VERSION'] = self.new_version subprocess.run([ 'docker-compose', '-f', f'docker-compose.{self.target_color}.yml', 'up', '-d' ], env=env, check=True) # Wait for deployment time.sleep(30) def health_check_target(self) -> bool: """Perform health check on target environment""" print(f"🏥 Health checking {self.target_color} environment") target_port = 5001 if self.target_color == 'green' else 5000 try: for attempt in range(10): response = requests.get(f'http://localhost:{target_port}/health', timeout=10) if response.status_code == 200: print(f"✅ {self.target_color} environment is healthy") return True time.sleep(10) print(f"❌ {self.target_color} environment failed health check") return False except Exception as e: print(f"❌ Health check failed: {str(e)}") return False def switch_traffic(self): """Switch traffic to target environment""" print(f"🔄 Switching traffic from {self.current_color} to {self.target_color}") # Update load balancer configuration # This would typically involve updating nginx config, cloud load balancer, etc. subprocess.run([ 'docker-compose', '-f', 'docker-compose.nginx.yml', 'restart' ], check=True) time.sleep(10) def verify_switch(self) -> bool: """Verify traffic switch was successful""" print("🔍 Verifying traffic switch") try: # Check main endpoint is responding response = requests.get('http://localhost/health', timeout=10) if response.status_code == 200: data = response.json() if data.get('version') == self.new_version: print("✅ Traffic switch verified") return True print("❌ Traffic switch verification failed") return False except Exception as e: print(f"❌ Traffic switch verification failed: {str(e)}") return False def cleanup_old_environment(self): """Clean up old environment""" print(f"🧹 Cleaning up {self.current_color} environment") subprocess.run([ 'docker-compose', '-f', f'docker-compose.{self.current_color}.yml', 'down' ], check=False) def rollback(self): """Rollback to previous environment""" print(f"🔄 Rolling back to {self.current_color} environment") # Switch traffic back subprocess.run([ 'docker-compose', '-f', 'docker-compose.nginx.yml', 'restart' ], check=True) # Clean up failed deployment subprocess.run([ 'docker-compose', '-f', f'docker-compose.{self.target_color}.yml', 'down' ], check=False) print(f"✅ Rollback to {self.current_color} completed") def deploy(self) -> bool: """Execute blue-green deployment""" try: print(f"🎯 Starting blue-green deployment") print(f"Current: {self.current_color}, Target: {self.target_color}") # Deploy to target environment self.deploy_target_environment() # Health check target if not self.health_check_target(): self.rollback() return False # Switch traffic self.switch_traffic() # Verify switch if not self.verify_switch(): self.rollback() return False # Cleanup old environment self.cleanup_old_environment() print("🎉 Blue-green deployment completed successfully!") return True except Exception as e: print(f"❌ Deployment failed: {str(e)}") self.rollback() return False def main(): if len(sys.argv) != 3: print("Usage: python blue_green_deploy.py ") sys.exit(1) app_name = sys.argv[1] version = sys.argv[2] deployment = BlueGreenDeployment(app_name, version) success = deployment.deploy() sys.exit(0 if success else 1) if __name__ == '__main__': main()