REL03

REL03-BP03 - Provide service contracts per API

REL03-BP03: Provide service contracts per API

Overview

Establish clear, well-defined service contracts for each API to ensure reliable communication between services and clients. Service contracts define the interface, data formats, error handling, versioning strategy, and behavioral expectations, enabling loose coupling, independent evolution, and reliable integration patterns across distributed systems.

Implementation Steps

1. Define Comprehensive API Specifications

Create detailed OpenAPI/Swagger specifications for all service endpoints
Define request and response schemas with validation rules
Specify error codes, messages, and handling patterns
Document authentication and authorization requirements

2. Implement Contract-First Development

Design API contracts before implementation begins
Use contract specifications to generate client SDKs and server stubs
Establish contract validation in CI/CD pipelines
Implement contract testing to ensure compliance

3. Establish API Versioning Strategy

Implement semantic versioning for API contracts
Design backward-compatible changes and deprecation policies
Provide multiple API versions simultaneously during transitions
Establish clear migration paths for breaking changes

4. Implement Contract Validation and Testing

Deploy contract testing frameworks for producer and consumer validation
Implement schema validation for all API requests and responses
Create automated tests that verify contract compliance
Establish contract regression testing in deployment pipelines

5. Design Error Handling and Resilience Patterns

Define standardized error response formats across all APIs
Implement circuit breaker patterns for service dependencies
Design retry policies and timeout configurations
Establish graceful degradation strategies for service failures

6. Establish Contract Governance and Evolution

Create processes for contract change management and approval
Implement contract versioning and lifecycle management
Establish deprecation policies and migration support
Maintain contract documentation and change logs

Implementation Examples

Example 1: API Contract Management and Validation System

{% raw %}

View code

import boto3
import json
import logging
import yaml
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any, Union
from dataclasses import dataclass, asdict, field
from enum import Enum
import jsonschema
from jsonschema import validate, ValidationError
import semver
import requests
from pathlib import Path

class ContractType(Enum):
    OPENAPI = "openapi"
    ASYNCAPI = "asyncapi"
    GRAPHQL = "graphql"
    GRPC = "grpc"

class VersioningStrategy(Enum):
    SEMANTIC = "semantic"
    DATE_BASED = "date_based"
    SEQUENTIAL = "sequential"

class CompatibilityLevel(Enum):
    BACKWARD = "backward"
    FORWARD = "forward"
    FULL = "full"
    BREAKING = "breaking"

@dataclass
class APIContract:
    contract_id: str
    service_name: str
    api_name: str
    version: str
    contract_type: ContractType
    specification: Dict[str, Any]
    created_at: str
    updated_at: str
    status: str
    compatibility_level: CompatibilityLevel
    deprecation_date: Optional[str] = None
    migration_guide: Optional[str] = None

@dataclass
class ContractValidationResult:
    is_valid: bool
    errors: List[str]
    warnings: List[str]
    compatibility_issues: List[str]
    validation_timestamp: str

class APIContractManager:
    def __init__(self, config: Dict):
        self.config = config
        self.s3 = boto3.client('s3')
        self.dynamodb = boto3.resource('dynamodb')
        self.apigateway = boto3.client('apigateway')
        self.cloudwatch = boto3.client('cloudwatch')
        self.sns = boto3.client('sns')
        
        # Initialize contract storage
        self.contracts_table = self.dynamodb.Table(
            config.get('contracts_table_name', 'api-contracts')
        )
        self.contracts_bucket = config.get('contracts_bucket_name', 'api-contracts-storage')
        
    def create_api_contract(self, contract_data: Dict) -> Dict:
        """Create and validate a new API contract"""
        contract_id = f"{contract_data['service_name']}_{contract_data['api_name']}_{contract_data['version']}"
        
        try:
            # Validate contract specification
            validation_result = self.validate_contract_specification(
                contract_data['specification'], 
                ContractType(contract_data['contract_type'])
            )
            
            if not validation_result.is_valid:
                raise ValueError(f"Contract validation failed: {validation_result.errors}")
            
            # Check version compatibility
            compatibility_result = self.check_version_compatibility(
                contract_data['service_name'],
                contract_data['api_name'],
                contract_data['version'],
                contract_data['specification']
            )
            
            # Create contract object
            contract = APIContract(
                contract_id=contract_id,
                service_name=contract_data['service_name'],
                api_name=contract_data['api_name'],
                version=contract_data['version'],
                contract_type=ContractType(contract_data['contract_type']),
                specification=contract_data['specification'],
                created_at=datetime.utcnow().isoformat(),
                updated_at=datetime.utcnow().isoformat(),
                status='active',
                compatibility_level=compatibility_result['compatibility_level'],
                deprecation_date=contract_data.get('deprecation_date'),
                migration_guide=contract_data.get('migration_guide')
            )
            
            # Store contract in DynamoDB
            self.contracts_table.put_item(Item=asdict(contract))
            
            # Store specification in S3
            self.store_contract_specification(contract_id, contract_data['specification'])
            
            # Generate client SDKs if requested
            if contract_data.get('generate_sdks', False):
                sdk_results = self.generate_client_sdks(contract)
                contract_dict = asdict(contract)
                contract_dict['sdk_results'] = sdk_results
                return contract_dict
            
            return asdict(contract)
            
        except Exception as e:
            logging.error(f"Failed to create API contract: {str(e)}")
            raise
    
    def validate_contract_specification(self, specification: Dict, contract_type: ContractType) -> ContractValidationResult:
        """Validate API contract specification"""
        errors = []
        warnings = []
        compatibility_issues = []
        
        try:
            if contract_type == ContractType.OPENAPI:
                errors.extend(self.validate_openapi_specification(specification))
            elif contract_type == ContractType.ASYNCAPI:
                errors.extend(self.validate_asyncapi_specification(specification))
            elif contract_type == ContractType.GRAPHQL:
                errors.extend(self.validate_graphql_specification(specification))
            
            # Check for common API design issues
            warnings.extend(self.check_api_design_patterns(specification, contract_type))
            
            # Check for breaking changes if this is an update
            compatibility_issues.extend(self.check_breaking_changes(specification, contract_type))
            
            return ContractValidationResult(
                is_valid=len(errors) == 0,
                errors=errors,
                warnings=warnings,
                compatibility_issues=compatibility_issues,
                validation_timestamp=datetime.utcnow().isoformat()
            )
            
        except Exception as e:
            logging.error(f"Contract validation failed: {str(e)}")
            return ContractValidationResult(
                is_valid=False,
                errors=[f"Validation error: {str(e)}"],
                warnings=[],
                compatibility_issues=[],
                validation_timestamp=datetime.utcnow().isoformat()
            )
    
    def validate_openapi_specification(self, specification: Dict) -> List[str]:
        """Validate OpenAPI specification"""
        errors = []
        
        try:
            # Check required OpenAPI fields
            required_fields = ['openapi', 'info', 'paths']
            for field in required_fields:
                if field not in specification:
                    errors.append(f"Missing required field: {field}")
            
            # Validate OpenAPI version
            if 'openapi' in specification:
                openapi_version = specification['openapi']
                if not openapi_version.startswith('3.'):
                    errors.append(f"Unsupported OpenAPI version: {openapi_version}")
            
            # Validate info section
            if 'info' in specification:
                info = specification['info']
                if 'title' not in info:
                    errors.append("Missing API title in info section")
                if 'version' not in info:
                    errors.append("Missing API version in info section")
            
            # Validate paths
            if 'paths' in specification:
                paths = specification['paths']
                if not paths:
                    errors.append("No paths defined in API specification")
                
                for path, path_item in paths.items():
                    if not isinstance(path_item, dict):
                        errors.append(f"Invalid path item for {path}")
                        continue
                    
                    # Check for HTTP methods
                    http_methods = ['get', 'post', 'put', 'delete', 'patch', 'head', 'options']
                    has_methods = any(method in path_item for method in http_methods)
                    if not has_methods:
                        errors.append(f"No HTTP methods defined for path {path}")
                    
                    # Validate each operation
                    for method, operation in path_item.items():
                        if method in http_methods:
                            if 'responses' not in operation:
                                errors.append(f"Missing responses for {method.upper()} {path}")
            
            # Validate components/schemas if present
            if 'components' in specification and 'schemas' in specification['components']:
                schemas = specification['components']['schemas']
                for schema_name, schema_def in schemas.items():
                    if not isinstance(schema_def, dict):
                        errors.append(f"Invalid schema definition for {schema_name}")
            
            return errors
            
        except Exception as e:
            return [f"OpenAPI validation error: {str(e)}"]
    
    def check_api_design_patterns(self, specification: Dict, contract_type: ContractType) -> List[str]:
        """Check for API design pattern compliance"""
        warnings = []
        
        try:
            if contract_type == ContractType.OPENAPI:
                # Check for RESTful design patterns
                if 'paths' in specification:
                    paths = specification['paths']
                    
                    # Check for proper HTTP method usage
                    for path, path_item in paths.items():
                        # Check for GET methods that modify state
                        if 'get' in path_item:
                            get_op = path_item['get']
                            if 'requestBody' in get_op:
                                warnings.append(f"GET method should not have request body: {path}")
                        
                        # Check for proper status codes
                        for method, operation in path_item.items():
                            if method in ['get', 'post', 'put', 'delete', 'patch']:
                                responses = operation.get('responses', {})
                                
                                # Check for success responses
                                success_codes = [code for code in responses.keys() if code.startswith('2')]
                                if not success_codes:
                                    warnings.append(f"No success response defined for {method.upper()} {path}")
                                
                                # Check for error responses
                                error_codes = [code for code in responses.keys() if code.startswith('4') or code.startswith('5')]
                                if not error_codes:
                                    warnings.append(f"No error responses defined for {method.upper()} {path}")
                
                # Check for consistent error response format
                if 'components' in specification and 'schemas' in specification['components']:
                    schemas = specification['components']['schemas']
                    error_schemas = [name for name in schemas.keys() if 'error' in name.lower()]
                    if not error_schemas:
                        warnings.append("No standardized error response schema found")
            
            return warnings
            
        except Exception as e:
            return [f"Design pattern check error: {str(e)}"]
    
    def generate_client_sdks(self, contract: APIContract) -> Dict[str, Any]:
        """Generate client SDKs from API contract"""
        sdk_results = {
            'generated_sdks': [],
            'generation_errors': [],
            'download_urls': {}
        }
        
        try:
            if contract.contract_type == ContractType.OPENAPI:
                # Generate Python SDK
                python_sdk = self.generate_python_sdk(contract)
                if python_sdk['success']:
                    sdk_results['generated_sdks'].append('python')
                    sdk_results['download_urls']['python'] = python_sdk['download_url']
                else:
                    sdk_results['generation_errors'].append(f"Python SDK: {python_sdk['error']}")
                
                # Generate JavaScript SDK
                js_sdk = self.generate_javascript_sdk(contract)
                if js_sdk['success']:
                    sdk_results['generated_sdks'].append('javascript')
                    sdk_results['download_urls']['javascript'] = js_sdk['download_url']
                else:
                    sdk_results['generation_errors'].append(f"JavaScript SDK: {js_sdk['error']}")
            
            return sdk_results
            
        except Exception as e:
            sdk_results['generation_errors'].append(f"SDK generation failed: {str(e)}")
            return sdk_results
    
    def generate_python_sdk(self, contract: APIContract) -> Dict[str, Any]:
        """Generate Python SDK from OpenAPI specification"""
        try:
            # Create Python client code
            client_code = f'''
"""
Generated Python SDK for {contract.service_name} {contract.api_name}
Version: {contract.version}
Generated at: {datetime.utcnow().isoformat()}
"""

import requests
import json
from typing import Dict, Any, Optional
from dataclasses import dataclass

@dataclass
class APIResponse:
    status_code: int
    data: Any
    headers: Dict[str, str]
    success: bool

class {contract.service_name.title()}{contract.api_name.title()}Client:
    """Client for {contract.service_name} {contract.api_name} API"""
    
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.session = requests.Session()
        
        if api_key:
            self.session.headers.update({{'Authorization': f'Bearer {{api_key}}'}})
    
    def _make_request(self, method: str, endpoint: str, **kwargs) -> APIResponse:
        """Make HTTP request to API"""
        url = f"{{self.base_url}}{{endpoint}}"
        
        try:
            response = self.session.request(method, url, **kwargs)
            
            # Parse JSON response if possible
            try:
                data = response.json()
            except:
                data = response.text
            
            return APIResponse(
                status_code=response.status_code,
                data=data,
                headers=dict(response.headers),
                success=200 <= response.status_code < 300
            )
            
        except Exception as e:
            return APIResponse(
                status_code=0,
                data={{'error': str(e)}},
                headers={{}},
                success=False
            )
'''
            
            # Add methods for each API endpoint
            if 'paths' in contract.specification:
                for path, path_item in contract.specification['paths'].items():
                    for method, operation in path_item.items():
                        if method in ['get', 'post', 'put', 'delete', 'patch']:
                            method_name = self.generate_method_name(method, path, operation)
                            method_code = self.generate_python_method(method, path, operation)
                            client_code += f"\n    {method_code}"
            
            # Store SDK in S3
            sdk_key = f"sdks/{contract.contract_id}/python/client.py"
            self.s3.put_object(
                Bucket=self.contracts_bucket,
                Key=sdk_key,
                Body=client_code,
                ContentType='text/x-python'
            )
            
            # Generate presigned URL for download
            download_url = self.s3.generate_presigned_url(
                'get_object',
                Params={'Bucket': self.contracts_bucket, 'Key': sdk_key},
                ExpiresIn=3600
            )
            
            return {
                'success': True,
                'download_url': download_url,
                'sdk_key': sdk_key
            }
            
        except Exception as e:
            return {
                'success': False,
                'error': str(e)
            }
    
    def generate_method_name(self, http_method: str, path: str, operation: Dict) -> str:
        """Generate Python method name from HTTP method and path"""
        # Use operationId if available
        if 'operationId' in operation:
            return operation['operationId']
        
        # Generate from HTTP method and path
        path_parts = [part for part in path.split('/') if part and not part.startswith('{')]
        method_name = http_method.lower()
        
        if path_parts:
            method_name += '_' + '_'.join(path_parts)
        
        return method_name.replace('-', '_')
    
    def generate_python_method(self, http_method: str, path: str, operation: Dict) -> str:
        """Generate Python method code for API operation"""
        method_name = self.generate_method_name(http_method, path, operation)
        
        # Extract parameters
        path_params = []
        query_params = []
        
        if 'parameters' in operation:
            for param in operation['parameters']:
                if param.get('in') == 'path':
                    path_params.append(param['name'])
                elif param.get('in') == 'query':
                    query_params.append(param['name'])
        
        # Generate method signature
        params = ['self'] + path_params
        if query_params:
            params.extend([f"{param}=None" for param in query_params])
        if http_method.lower() in ['post', 'put', 'patch']:
            params.append('data=None')
        
        method_signature = f"def {method_name}({', '.join(params)}) -> APIResponse:"
        
        # Generate method body
        method_body = f'''
        """
        {operation.get('summary', f'{http_method.upper()} {path}')}
        """
        endpoint = "{path}"
        '''
        
        # Replace path parameters
        for param in path_params:
            method_body += f'\n        endpoint = endpoint.replace("{{{param}}}", str({param}))'
        
        # Add query parameters
        if query_params:
            method_body += '\n        params = {}'
            for param in query_params:
                method_body += f'\n        if {param} is not None: params["{param}"] = {param}'
        
        # Make request
        request_args = [f'"{http_method.upper()}"', 'endpoint']
        if query_params:
            request_args.append('params=params')
        if http_method.lower() in ['post', 'put', 'patch']:
            request_args.append('json=data')
        
        method_body += f'\n        return self._make_request({", ".join(request_args)})'
        
        return method_signature + method_body
    
    def store_contract_specification(self, contract_id: str, specification: Dict):
        """Store contract specification in S3"""
        try:
            spec_key = f"contracts/{contract_id}/specification.json"
            self.s3.put_object(
                Bucket=self.contracts_bucket,
                Key=spec_key,
                Body=json.dumps(specification, indent=2),
                ContentType='application/json'
            )
            
        except Exception as e:
            logging.error(f"Failed to store contract specification: {str(e)}")
            raise

{% endraw %}

Example 2: API Contract Testing and Validation Script