API Lifecycle

# API design specification template
api_specification:
  name: "User Management API"
  version: "1.0.0"
  purpose: "Manage user accounts and profiles"
  
  business_requirements:
    - "Create and manage user accounts"
    - "Support user authentication"
    - "Enable profile management"
    - "Provide user search capabilities"
  
  technical_requirements:
    performance:
      response_time: "< 200ms for 95th percentile"
      throughput: "1000 requests/second"
      availability: "99.9% uptime"
    
    security:
      authentication: "JWT tokens"
      authorization: "Role-based access control"
      encryption: "TLS 1.3 in transit, AES-256 at rest"
    
    scalability:
      horizontal_scaling: true
      load_balancing: true
      caching: "Redis-based"
  
  data_model:
    User:
      properties:
        - id: string (UUID)
        - email: string (unique)
        - name: string
        - role: enum [admin, user]
        - created_at: datetime
        - updated_at: datetime
  
  endpoints:
    - path: "/users"
      methods: [GET, POST]
      purpose: "List and create users"
    - path: "/users/{id}"
      methods: [GET, PUT, DELETE]
      purpose: "Manage individual users"

# Development structure following lifecycle best practices
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr
from sqlalchemy.orm import Session
import logging
 
# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
 
app = FastAPI(
    title="User Management API",
    version="1.0.0",
    description="API for managing user accounts and profiles",
    docs_url="/docs",
    redoc_url="/redoc"
)
 
# Data models with validation
class UserCreate(BaseModel):
    email: EmailStr
    name: str
    role: str = "user"
 
class UserResponse(BaseModel):
    id: str
    email: str
    name: str
    role: str
    created_at: str
    updated_at: str
    
    class Config:
        orm_mode = True
 
# Dependency injection for database sessions
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
# Health check endpoint (required for lifecycle management)
@app.get("/health")
async def health_check():
    """Health check endpoint for monitoring and deployment."""
    return {
        "status": "healthy",
        "version": "1.0.0",
        "timestamp": datetime.utcnow().isoformat()
    }
 
# Version endpoint
@app.get("/version")
async def get_version():
    """Get API version information."""
    return {
        "name": "User Management API",
        "version": "1.0.0",
        "build_date": "2024-01-15",
        "git_commit": "abc123def456"
    }
 
# Main API endpoints with proper error handling and logging
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(
    user_data: UserCreate,
    db: Session = Depends(get_db)
):
    """Create a new user account."""
    try:
        logger.info(f"Creating user with email: {user_data.email}")
        
        # Check if user already exists
        existing_user = db.query(User).filter(User.email == user_data.email).first()
        if existing_user:
            raise HTTPException(status_code=409, detail="User already exists")
        
        # Create new user
        new_user = User(**user_data.dict())
        db.add(new_user)
        db.commit()
        db.refresh(new_user)
        
        logger.info(f"User created successfully: {new_user.id}")
        return new_user
        
    except Exception as e:
        logger.error(f"Error creating user: {str(e)}")
        db.rollback()
        raise HTTPException(status_code=500, detail="Internal server error")

# Comprehensive testing approach for lifecycle management
import pytest
from fastapi.testclient import TestClient
from unittest.mock import Mock, patch
import json
 
class TestUserAPI:
    """Test suite covering all lifecycle phases."""
    
    def setup_method(self):
        """Setup test environment."""
        self.client = TestClient(app)
        self.test_user_data = {
            "email": "test@example.com",
            "name": "Test User",
            "role": "user"
        }
    
    def test_health_check(self):
        """Verify health check endpoint works (deployment requirement)."""
        response = self.client.get("/health")
        assert response.status_code == 200
        data = response.json()
        assert data["status"] == "healthy"
        assert "version" in data
    
    def test_version_endpoint(self):
        """Verify version information is available (lifecycle tracking)."""
        response = self.client.get("/version")
        assert response.status_code == 200
        data = response.json()
        assert "version" in data
        assert "name" in data
    
    def test_create_user_success(self):
        """Test successful user creation."""
        with patch('app.get_db') as mock_db:
            mock_session = Mock()
            mock_db.return_value = mock_session
            mock_session.query().filter().first.return_value = None
            
            response = self.client.post("/users", json=self.test_user_data)
            assert response.status_code == 201
    
    def test_create_user_duplicate_email(self):
        """Test duplicate email handling."""
        with patch('app.get_db') as mock_db:
            mock_session = Mock()
            mock_db.return_value = mock_session
            mock_session.query().filter().first.return_value = Mock()  # Existing user
            
            response = self.client.post("/users", json=self.test_user_data)
            assert response.status_code == 409
    
    def test_api_contract_compliance(self):
        """Verify API responses match specification."""
        # Test response structure
        with patch('app.get_db') as mock_db:
            mock_session = Mock()
            mock_user = Mock()
            mock_user.id = "123"
            mock_user.email = "test@example.com"
            mock_user.name = "Test User"
            mock_user.role = "user"
            mock_user.created_at = "2024-01-01T00:00:00Z"
            mock_user.updated_at = "2024-01-01T00:00:00Z"
            
            mock_session.add.return_value = None
            mock_session.commit.return_value = None
            mock_session.refresh.return_value = None
            mock_session.query().filter().first.return_value = None
            mock_db.return_value = mock_session
            
            # Mock the created user
            with patch('app.User') as mock_user_class:
                mock_user_class.return_value = mock_user
                
                response = self.client.post("/users", json=self.test_user_data)
                data = response.json()
                
                # Verify required fields are present
                required_fields = ["id", "email", "name", "role", "created_at", "updated_at"]
                for field in required_fields:
                    assert field in data

# .github/workflows/api-lifecycle.yml
name: API Lifecycle Management
 
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
 
env:
  API_VERSION: [mathematical expression]{{ github.sha }}
 
jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.9'
    
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov
    
    - name: Run tests with coverage
      run: |
        pytest --cov=src --cov-report=xml --cov-report=html
    
    - name: Upload coverage reports
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml
    
    - name: Validate API specification
      run: |
        swagger-codegen validate -i openapi.yaml
 
  build:
    needs: test
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Build Docker image
      run: |
        docker build -t api:{{ env.API_VERSION }} .
        docker tag api:{{ env.API_VERSION }} api:latest
    
    - name: Run security scan
      uses: aquasecurity/trivy-action@master
      with:
        image-ref: 'api:{{ env.API_VERSION }}'
        format: 'sarif'
        output: 'trivy-results.sarif'
    
    - name: Upload security scan results
      uses: github/codeql-action/upload-sarif@v2
      with:
        sarif_file: 'trivy-results.sarif'
 
  deploy-staging:
    needs: build
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/develop'
    
    steps:
    - name: Deploy to staging
      run: |
        echo "Deploying to staging environment"
        # Deployment script here
    
    - name: Run integration tests
      run: |
        pytest tests/integration/
    
    - name: Performance testing
      run: |
        docker run --rm -i loadimpact/k6 run - <tests/performance/load-test.js
 
  deploy-production:
    needs: [build, deploy-staging]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
    - name: Blue-green deployment
      run: |
        echo "Deploying to production with blue-green strategy"
        # Blue-green deployment script
    
    - name: Smoke tests
      run: |
        pytest tests/smoke/
    
    - name: Update API documentation
      run: |
        # Generate and deploy documentation
        redoc-cli build openapi.yaml --output docs/index.html

# Blue-green deployment script
import requests
import time
import sys
 
class BlueGreenDeployment:
    def __init__(self, current_env, new_env, health_check_url):
        self.current_env = current_env
        self.new_env = new_env
        self.health_check_url = health_check_url
    
    def deploy(self):
        """Execute blue-green deployment."""
        print(f"Starting deployment from {self.current_env} to {self.new_env}")
        
        # 1. Deploy to new environment
        if not self.deploy_new_version():
            print("Deployment failed")
            return False
        
        # 2. Health check new environment
        if not self.health_check():
            print("Health check failed")
            return False
        
        # 3. Switch traffic
        if not self.switch_traffic():
            print("Traffic switch failed")
            return False
        
        # 4. Verify switch success
        if not self.verify_switch():
            print("Switch verification failed")
            self.rollback()
            return False
        
        print("Deployment successful")
        return True
    
    def health_check(self, max_retries=5):
        """Check health of new environment."""
        for attempt in range(max_retries):
            try:
                response = requests.get(f"{self.new_env}{self.health_check_url}")
                if response.status_code == 200:
                    data = response.json()
                    if data.get('status') == 'healthy':
                        print(f"Health check passed on attempt {attempt + 1}")
                        return True
            except requests.RequestException as e:
                print(f"Health check attempt {attempt + 1} failed: {e}")
            
            time.sleep(10)  # Wait before retry
        
        return False
    
    def deploy_new_version(self):
        """Deploy new version to inactive environment."""
        # Implementation depends on deployment platform
        # Kubernetes, Docker Swarm, etc.
        return True
    
    def switch_traffic(self):
        """Switch load balancer traffic to new environment."""
        # Implementation depends on load balancer
        # HAProxy, nginx, cloud load balancer, etc.
        return True
    
    def rollback(self):
        """Rollback to previous version."""
        print("Rolling back deployment")
        # Implementation for rollback

# Performance monitoring and alerting
from dataclasses import dataclass
from typing import Dict, List
import time
import json
 
@dataclass
class PerformanceMetric:
    name: str
    value: float
    threshold: float
    unit: str
    timestamp: str
 
class APIPerformanceMonitor:
    def __init__(self):
        self.metrics = {}
        self.thresholds = {
            'response_time_95th': 200,  # ms
            'error_rate': 1.0,          # percentage
            'throughput': 100,          # requests/second
            'cpu_usage': 80,            # percentage
            'memory_usage': 80          # percentage
        }
    
    def collect_metrics(self) -> List[PerformanceMetric]:
        """Collect current performance metrics."""
        metrics = []
        
        # Response time (95th percentile)
        response_time = self.get_95th_percentile_response_time()
        metrics.append(PerformanceMetric(
            name='response_time_95th',
            value=response_time,
            threshold=self.thresholds['response_time_95th'],
            unit='ms',
            timestamp=datetime.utcnow().isoformat()
        ))
        
        # Error rate
        error_rate = self.get_error_rate()
        metrics.append(PerformanceMetric(
            name='error_rate',
            value=error_rate,
            threshold=self.thresholds['error_rate'],
            unit='percentage',
            timestamp=datetime.utcnow().isoformat()
        ))
        
        return metrics
    
    def check_thresholds(self, metrics: List[PerformanceMetric]) -> List[str]:
        """Check metrics against thresholds and return alerts."""
        alerts = []
        
        for metric in metrics:
            if metric.value > metric.threshold:
                alerts.append(
                    f"ALERT: {metric.name} is {metric.value}{metric.unit}, "
                    f"exceeding threshold of {metric.threshold}{metric.unit}"
                )
        
        return alerts

# Version management implementation
from enum import Enum
from dataclasses import dataclass
from typing import List, Optional
 
class ChangeType(Enum):
    BREAKING = "breaking"
    FEATURE = "feature"
    BUGFIX = "bugfix"
    SECURITY = "security"
 
@dataclass
class VersionChange:
    type: ChangeType
    description: str
    migration_required: bool = False
 
class APIVersionManager:
    def __init__(self):
        self.current_version = "1.0.0"
        self.supported_versions = ["1.0.0", "1.1.0", "2.0.0"]
        self.deprecated_versions = ["0.9.0"]
        self.changelog = {}
    
    def calculate_next_version(self, changes: List[VersionChange]) -> str:
        """Calculate next version based on changes using semantic versioning."""
        major, minor, patch = map(int, self.current_version.split('.'))
        
        has_breaking = any(change.type == ChangeType.BREAKING for change in changes)
        has_feature = any(change.type == ChangeType.FEATURE for change in changes)
        has_bugfix = any(change.type in [ChangeType.BUGFIX, ChangeType.SECURITY] for change in changes)
        
        if has_breaking:
            major += 1
            minor = 0
            patch = 0
        elif has_feature:
            minor += 1
            patch = 0
        elif has_bugfix:
            patch += 1
        
        return f"{major}.{minor}.{patch}"
    
    def deprecate_version(self, version: str, sunset_date: str):
        """Mark a version as deprecated with sunset date."""
        self.deprecated_versions.append(version)
        
        # Send deprecation notices
        self.send_deprecation_notice(version, sunset_date)
    
    def send_deprecation_notice(self, version: str, sunset_date: str):
        """Send deprecation notice to API consumers."""
        notice = {
            "type": "deprecation_notice",
            "deprecated_version": version,
            "sunset_date": sunset_date,
            "migration_guide": f"https://docs.example.com/migration/{version}",
            "support_contact": "api-support@example.com"
        }
        
        # Implementation to notify consumers
        print(f"Deprecation notice sent for version {version}")
 
# Version header middleware
from fastapi import Request, Response
 
@app.middleware("http")
async def version_middleware(request: Request, call_next):
    """Handle API versioning through headers."""
    requested_version = request.headers.get("API-Version", "latest")
    
    if requested_version == "latest":
        requested_version = API_VERSION_MANAGER.current_version
    
    # Check if version is supported
    if requested_version not in API_VERSION_MANAGER.supported_versions:
        return JSONResponse(
            status_code=400,
            content={
                "error": f"Unsupported API version: {requested_version}",
                "supported_versions": API_VERSION_MANAGER.supported_versions
            }
        )
    
    # Check if version is deprecated
    if requested_version in API_VERSION_MANAGER.deprecated_versions:
        response = await call_next(request)
        response.headers["Deprecation"] = "true"
        response.headers["Sunset"] = "2024-12-31"  # Sunset date
        return response
    
    response = await call_next(request)
    response.headers["API-Version"] = requested_version
    return response

# API retirement management
from datetime import datetime, timedelta
import logging
 
class APIRetirementManager:
    def __init__(self):
        self.retirement_schedule = {}
        self.logger = logging.getLogger(__name__)
    
    def schedule_retirement(self, version: str, retirement_date: str, 
                          migration_guide_url: str):
        """Schedule API version retirement."""
        self.retirement_schedule[version] = {
            "retirement_date": retirement_date,
            "migration_guide": migration_guide_url,
            "notification_sent": False,
            "final_warning_sent": False
        }
        
        # Send initial retirement notification
        self.send_retirement_notification(version, retirement_date, migration_guide_url)
    
    def check_retirement_status(self):
        """Check and update retirement status for all scheduled versions."""
        current_date = datetime.now()
        
        for version, schedule in self.retirement_schedule.items():
            retirement_date = datetime.fromisoformat(schedule["retirement_date"])
            days_until_retirement = (retirement_date - current_date).days
            
            # Final warning 30 days before retirement
            if days_until_retirement <= 30 and not schedule["final_warning_sent"]:
                self.send_final_warning(version, days_until_retirement)
                schedule["final_warning_sent"] = True
            
            # Execute retirement
            if days_until_retirement <= 0:
                self.execute_retirement(version)
    
    def execute_retirement(self, version: str):
        """Execute API version retirement."""
        self.logger.info(f"Retiring API version {version}")
        
        # Remove from supported versions
        if version in API_VERSION_MANAGER.supported_versions:
            API_VERSION_MANAGER.supported_versions.remove(version)
        
        # Update load balancer to reject requests to this version
        self.update_load_balancer_config(version)
        
        # Archive documentation
        self.archive_documentation(version)
        
        self.logger.info(f"API version {version} successfully retired")
    
    def send_retirement_notification(self, version: str, retirement_date: str, 
                                   migration_guide: str):
        """Send initial retirement notification."""
        message = {
            "type": "retirement_notice",
            "version": version,
            "retirement_date": retirement_date,
            "migration_guide": migration_guide,
            "contact": "api-support@example.com"
        }
        
        # Implementation to notify all consumers
        self.logger.info(f"Retirement notice sent for version {version}")

# Terraform configuration for API lifecycle infrastructure
terraform {
  required_version = ">= 1.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}
 
resource "aws_api_gateway_rest_api" "user_api" {
  name        = "user-management-api"
  description = "User Management API - Lifecycle Managed"
  
  endpoint_configuration {
    types = ["REGIONAL"]
  }
  
  lifecycle {
    create_before_destroy = true
  }
  
  tags = {
    Environment = var.environment
    Version     = var.api_version
    Lifecycle   = "managed"
  }
}
 
resource "aws_api_gateway_deployment" "api_deployment" {
  depends_on = [aws_api_gateway_rest_api.user_api]
  
  rest_api_id = aws_api_gateway_rest_api.user_api.id
  stage_name  = var.environment
  
  lifecycle {
    create_before_destroy = true
  }
  
  variables = {
    deployed_at = timestamp()
    version     = var.api_version
  }
}
 
# Auto-scaling for different lifecycle phases
resource "aws_autoscaling_group" "api_servers" {
  name                = "api-servers-{var.environment}"
  vpc_zone_identifier = var.subnet_ids
  target_group_arns   = [aws_lb_target_group.api_tg.arn]
  health_check_type   = "ELB"
  
  min_size         = var.environment == "production" ? 3 : 1
  max_size         = var.environment == "production" ? 10 : 3
  desired_capacity = var.environment == "production" ? 3 : 1
  
  tag {
    key                 = "Name"
    value               = "api-server-{var.environment}"
    propagate_at_launch = true
  }
  
  tag {
    key                 = "Lifecycle"
    value               = var.environment
    propagate_at_launch = true
  }
}