API Management Tools
API management tools provide the infrastructure and capabilities needed to design, deploy, secure, monitor, and maintain APIs effectively. These tools streamline the API lifecycle and enable teams to focus on business logic rather than infrastructure concerns.
API Gateway Solutions
Kong Gateway
Kong is a cloud-native API gateway built for hybrid and multi-cloud environments.
Configuration Example
# Kong Gateway configuration
# docker-compose.yml
version: '3.8'
services:
kong:
image: kong:latest
environment:
- KONG_DATABASE=postgres
- KONG_PG_HOST=postgres
- KONG_PG_DATABASE=kong
- KONG_PG_USER=kong
- KONG_PG_PASSWORD=kong
- KONG_PROXY_ACCESS_LOG=/dev/stdout
- KONG_ADMIN_ACCESS_LOG=/dev/stdout
- KONG_PROXY_ERROR_LOG=/dev/stderr
- KONG_ADMIN_ERROR_LOG=/dev/stderr
- KONG_ADMIN_LISTEN=0.0.0.0:8001
ports:
- "8000:8000" # Proxy
- "8001:8001" # Admin API
depends_on:
- postgres
postgres:
image: postgres:13
environment:
POSTGRES_DB: kong
POSTGRES_USER: kong
POSTGRES_PASSWORD: kong
volumes:
- postgres_data:/var/lib/postgresql/data
volumes:
postgres_data:
# Kong service and route configuration
apiVersion: v1
kind: ConfigMap
metadata:
name: kong-config
data:
kong.yml: |
_format_version: "3.0"
services:
- name: user-api
url: http://user-service:8080
tags:
- production
routes:
- name: user-routes
paths:
- /api/v1/users
methods:
- GET
- POST
- PUT
- DELETE
strip_path: false
plugins:
- name: rate-limiting
config:
minute: 100
hour: 1000
- name: key-auth
config:
key_names:
- X-API-Key
- name: cors
config:
origins:
- "https://dashboard.example.com"
- "https://app.example.com"
methods:
- GET
- POST
- PUT
- DELETE
- OPTIONS
headers:
- Accept
- Content-Type
- Authorization
- X-API-Key
Kong Plugin Development
-- Custom Kong plugin for API analytics
-- plugins/api-analytics/handler.lua
local BasePlugin = require "kong.plugins.base_plugin"
local json = require "cjson"
local APIAnalyticsHandler = BasePlugin:extend()
APIAnalyticsHandler.PRIORITY = 1000
APIAnalyticsHandler.VERSION = "1.0.0"
function APIAnalyticsHandler:new()
APIAnalyticsHandler.super.new(self, "api-analytics")
end
function APIAnalyticsHandler:access(conf)
APIAnalyticsHandler.super.access(self)
-- Capture request start time
kong.ctx.shared.start_time = kong.request.get_start_time()
-- Extract client information
local client_id = kong.request.get_header("X-API-Key") or "anonymous"
local user_agent = kong.request.get_header("User-Agent") or "unknown"
-- Store in context for response phase
kong.ctx.shared.client_id = client_id
kong.ctx.shared.user_agent = user_agent
end
function APIAnalyticsHandler:log(conf)
APIAnalyticsHandler.super.log(self)
-- Calculate response time
local start_time = kong.ctx.shared.start_time
local response_time = kong.request.get_start_time() - start_time
-- Collect analytics data
local analytics_data = {
timestamp = os.time(),
client_id = kong.ctx.shared.client_id,
method = kong.request.get_method(),
path = kong.request.get_path(),
status_code = kong.response.get_status(),
response_time = response_time,
user_agent = kong.ctx.shared.user_agent,
request_size = kong.request.get_size(),
response_size = kong.response.get_size()
}
-- Send to analytics service (Redis, Kafka, etc.)
local redis = require "resty.redis"
local red = redis:new()
local ok, err = red:connect("redis-host", 6379)
if ok then
red:lpush("api_analytics", json.encode(analytics_data))
red:expire("api_analytics", 86400) -- Expire after 24 hours
end
kong.log.info("Analytics data collected: ", json.encode(analytics_data))
end
return APIAnalyticsHandler
-- plugins/api-analytics/schema.lua
return {
name = "api-analytics",
fields = {
{ config = {
type = "record",
fields = {
{ redis_host = { type = "string", default = "localhost" } },
{ redis_port = { type = "number", default = 6379 } },
{ redis_timeout = { type = "number", default = 1000 } },
}
}
}
}
}
AWS API Gateway
AWS API Gateway provides managed API hosting with built-in security and monitoring.
Terraform Configuration
# AWS API Gateway with Lambda integration
resource "aws_api_gateway_rest_api" "user_api" {
name = "user-management-api"
description = "User Management API with AWS API Gateway"
endpoint_configuration {
types = ["REGIONAL"]
}
tags = {
Environment = var.environment
Team = "platform"
}
}
resource "aws_api_gateway_resource" "users" {
rest_api_id = aws_api_gateway_rest_api.user_api.id
parent_id = aws_api_gateway_rest_api.user_api.root_resource_id
path_part = "users"
}
resource "aws_api_gateway_method" "get_users" {
rest_api_id = aws_api_gateway_rest_api.user_api.id
resource_id = aws_api_gateway_resource.users.id
http_method = "GET"
authorization = "AWS_IAM"
request_parameters = {
"method.request.querystring.limit" = false
"method.request.querystring.offset" = false
"method.request.header.X-API-Key" = true
}
}
resource "aws_api_gateway_integration" "users_lambda" {
rest_api_id = aws_api_gateway_rest_api.user_api.id
resource_id = aws_api_gateway_resource.users.id
http_method = aws_api_gateway_method.get_users.http_method
integration_http_method = "POST"
type = "AWS_PROXY"
uri = aws_lambda_function.user_handler.invoke_arn
}
# Request/Response transformation
resource "aws_api_gateway_method_response" "users_200" {
rest_api_id = aws_api_gateway_rest_api.user_api.id
resource_id = aws_api_gateway_resource.users.id
http_method = aws_api_gateway_method.get_users.http_method
status_code = "200"
response_headers = {
"Access-Control-Allow-Origin" = true
"Access-Control-Allow-Headers" = true
"Access-Control-Allow-Methods" = true
"X-Rate-Limit-Remaining" = true
}
response_models = {
"application/json" = aws_api_gateway_model.user_list.name
}
}
resource "aws_api_gateway_model" "user_list" {
rest_api_id = aws_api_gateway_rest_api.user_api.id
name = "UserList"
content_type = "application/json"
schema = jsonencode({
"schema" = "http://json-schema.org/draft-04/schema#"
title = "User List Response"
type = "object"
properties = {
data = {
type = "array"
items = {
"ref" = "#/definitions/User"
}
}
meta = {
"ref" = "#/definitions/ResponseMeta"
}
}
definitions = {
User = {
type = "object"
properties = {
id = { type = "string" }
email = { type = "string" }
name = { type = "string" }
}
}
ResponseMeta = {
type = "object"
properties = {
total = { type = "number" }
timestamp = { type = "string" }
}
}
}
})
}
# Usage plan and API keys
resource "aws_api_gateway_usage_plan" "user_api_plan" {
name = "user-api-standard-plan"
api_stages {
api_id = aws_api_gateway_rest_api.user_api.id
stage = aws_api_gateway_deployment.production.stage_name
}
quota_settings {
limit = 10000
period = "MONTH"
}
throttle_settings {
rate_limit = 100
burst_limit = 200
}
}
resource "aws_api_gateway_api_key" "client_key" {
name = "user-api-client-key"
tags = {
Client = "web-dashboard"
}
}
resource "aws_api_gateway_usage_plan_key" "main" {
key_id = aws_api_gateway_api_key.client_key.id
key_type = "API_KEY"
usage_plan_id = aws_api_gateway_usage_plan.user_api_plan.id
}
Documentation Tools
Swagger/OpenAPI Tools
Swagger UI with Custom Theming
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Data Engineering API Documentation</title>
<link rel="stylesheet" type="text/css" href="https://unpkg.com/swagger-ui-dist@4.15.5/swagger-ui.css" />
<style>
/* Custom API documentation styling */
.swagger-ui {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
}
.swagger-ui .topbar {
background-color: #1a365d;
border-bottom: 3px solid #2d3748;
}
.swagger-ui .topbar .download-url-wrapper {
display: none;
}
.swagger-ui .info {
margin: 50px 0;
}
.swagger-ui .info .title {
color: #1a365d;
font-size: 2.5em;
}
.swagger-ui .scheme-container {
background: #f7fafc;
border: 1px solid #e2e8f0;
border-radius: 8px;
padding: 20px;
margin: 20px 0;
}
.swagger-ui .opblock.opblock-get {
border-color: #38a169;
background: rgba(56, 161, 105, 0.1);
}
.swagger-ui .opblock.opblock-post {
border-color: #3182ce;
background: rgba(49, 130, 206, 0.1);
}
.swagger-ui .opblock.opblock-put {
border-color: #d69e2e;
background: rgba(214, 158, 46, 0.1);
}
.swagger-ui .opblock.opblock-delete {
border-color: #e53e3e;
background: rgba(229, 62, 62, 0.1);
}
.custom-logo {
max-width: 200px;
margin-bottom: 20px;
}
.api-stats {
display: flex;
justify-content: space-around;
background: #f7fafc;
padding: 20px;
border-radius: 8px;
margin: 20px 0;
}
.stat-item {
text-align: center;
}
.stat-number {
font-size: 2em;
font-weight: bold;
color: #1a365d;
}
.stat-label {
color: #4a5568;
font-size: 0.9em;
}
</style>
</head>
<body>
<div id="swagger-ui"></div>
<script src="https://unpkg.com/swagger-ui-dist@4.15.5/swagger-ui-bundle.js"></script>
<script src="https://unpkg.com/swagger-ui-dist@4.15.5/swagger-ui-standalone-preset.js"></script>
<script>
// Custom plugin for additional features
const CustomPlugin = function() {
return {
components: {
// Custom logo component
Logo: () => {
return React.createElement('div', {
className: 'custom-logo-container'
}, [
React.createElement('img', {
src: '/assets/api-logo.png',
alt: 'API Logo',
className: 'custom-logo'
}),
React.createElement('div', {
className: 'api-stats'
}, [
React.createElement('div', { className: 'stat-item' }, [
React.createElement('div', { className: 'stat-number' }, '25'),
React.createElement('div', { className: 'stat-label' }, 'Endpoints')
]),
React.createElement('div', { className: 'stat-item' }, [
React.createElement('div', { className: 'stat-number' }, '99.9%'),
React.createElement('div', { className: 'stat-label' }, 'Uptime')
]),
React.createElement('div', { className: 'stat-item' }, [
React.createElement('div', { className: 'stat-number' }, '150ms'),
React.createElement('div', { className: 'stat-label' }, 'Avg Response')
])
])
]);
}
},
wrapComponents: {
// Wrap info component to add custom content
info: (Original, system) => (props) => {
return React.createElement('div', null, [
React.createElement(Original, props),
React.createElement('div', {
style: {
background: '#e6fffa',
border: '1px solid #81e6d9',
borderRadius: '8px',
padding: '16px',
margin: '20px 0'
}
}, [
React.createElement('h3', {
style: { margin: '0 0 10px 0', color: '#234e52' }
}, 'Quick Start'),
React.createElement('p', {
style: { margin: 0, color: '#285e61' }
}, 'Get your API key from the developer portal and start making requests in minutes!')
])
]);
}
}
};
};
// Initialize Swagger UI
window.ui = SwaggerUIBundle({
url: '/openapi.json',
dom_id: '#swagger-ui',
deepLinking: true,
presets: [
SwaggerUIBundle.presets.apis,
SwaggerUIBundle.presets.standalone
],
plugins: [
SwaggerUIBundle.plugins.DownloadUrl,
CustomPlugin
],
layout: "StandaloneLayout",
requestInterceptor: (request) => {
// Add API key to all requests in documentation
if (!request.headers['X-API-Key']) {
request.headers['X-API-Key'] = 'demo-api-key-for-docs';
}
return request;
},
responseInterceptor: (response) => {
// Log response for debugging
console.log('API Response:', response);
return response;
},
onComplete: () => {
console.log('Swagger UI loaded successfully');
},
tryItOutEnabled: true,
filter: true,
supportedSubmitMethods: ['get', 'post', 'put', 'delete', 'patch'],
validatorUrl: null // Disable online validator
});
</script>
</body>
</html>
Postman Integration
Automated Collection Generation
// Generate Postman collection from OpenAPI spec
const fs = require('fs');
const yaml = require('js-yaml');
const { Converter } = require('openapi-to-postmanv2');
class PostmanCollectionGenerator {
constructor(openApiPath, outputPath) {
this.openApiPath = openApiPath;
this.outputPath = outputPath;
}
async generateCollection() {
try {
// Read OpenAPI specification
const openApiContent = fs.readFileSync(this.openApiPath, 'utf8');
const openApiSpec = yaml.load(openApiContent);
// Convert to Postman collection
const converter = new Converter();
const conversionResult = await new Promise((resolve, reject) => {
converter.convert({
type: 'json',
data: openApiSpec
}, {
// Conversion options
requestParametersResolution: 'Example',
exampleParametersResolution: 'Example',
folderStrategy: 'Tags',
requestNameSource: 'Fallback',
indentCharacter: ' ',
collectionName: 'Data Engineering API Collection',
includeAuthInfoInExample: true,
enableOptionalParameters: true
}, (err, result) => {
if (err) reject(err);
else resolve(result);
});
});
if (!conversionResult.result) {
throw new Error('Conversion failed');
}
const collection = conversionResult.output[0].data;
// Enhance collection with custom features
this.enhanceCollection(collection);
// Write to file
fs.writeFileSync(
this.outputPath,
JSON.stringify(collection, null, 2)
);
console.log(`Postman collection generated: {this.outputPath}`);
} catch (error) {
console.error('Error generating Postman collection:', error);
throw error;
}
}
enhanceCollection(collection) {
// Add collection-level auth
collection.auth = {
type: 'apikey',
apikey: [
{
key: 'key',
value: 'X-API-Key',
type: 'string'
},
{
key: 'value',
value: '{{api_key}}',
type: 'string'
}
]
};
// Add environment variables
collection.variable = [
{
key: 'base_url',
value: 'https://api.example.com/v2',
type: 'string'
},
{
key: 'api_key',
value: 'your-api-key-here',
type: 'string'
}
];
// Add pre-request scripts and tests to all items
this.addScriptsToItems(collection.item);
// Add collection info
collection.info.description = `
# Data Engineering API Collection
This Postman collection provides complete examples for interacting with the Data Engineering API.
## Setup Instructions
1. Set your API key in the \`api_key\` variable
2. Update the \`base_url\` if using a different environment
3. Run the requests in the "Authentication" folder first
## Environment Variables
- \`base_url\`: API base URL
- \`api_key\`: Your API authentication key
- \`user_id\`: Current user ID (set automatically)
- \`access_token\`: JWT access token (set automatically)
## Test Automation
Each request includes automated tests that:
- Verify response status codes
- Validate response structure
- Extract values for subsequent requests
- Check response times
`;
}
addScriptsToItems(items) {
items.forEach(item => {
if (item.request) {
// Add pre-request script
item.event = item.event || [];
// Pre-request script
item.event.push({
listen: 'prerequest',
script: {
exec: [
'// Set timestamp for request tracking',
'pm.environment.set("timestamp", new Date().toISOString());',
'',
'// Log request details',
'console.log(`Making [mathematical expression]{pm.request.url}`);',
'',
'// Add request ID for tracing',
'const requestId = "req_" + Math.random().toString(36).substring(2, 15);',
'pm.request.headers.add({',
' key: "X-Request-ID",',
' value: requestId',
'});'
]
}
});
// Test script
item.event.push({
listen: 'test',
script: {
exec: [
'// Basic response validation',
'pm.test("Response status code is successful", function () {',
' pm.expect(pm.response.code).to.be.oneOf([200, 201, 202, 204]);',
'});',
'',
'pm.test("Response time is acceptable", function () {',
' pm.expect(pm.response.responseTime).to.be.below(2000);',
'});',
'',
'pm.test("Response has correct content type", function () {',
' pm.expect(pm.response.headers.get("Content-Type")).to.include("application/json");',
'});',
'',
'// Extract useful values for subsequent requests',
'if (pm.response.code === 200 || pm.response.code === 201) {',
' try {',
' const jsonData = pm.response.json();',
' ',
' // Extract user ID if present',
' if (jsonData.id) {',
' pm.environment.set("user_id", jsonData.id);',
' }',
' ',
' // Extract access token if present',
' if (jsonData.access_token) {',
' pm.environment.set("access_token", jsonData.access_token);',
' }',
' ',
' // Log response summary',
' console.log(`Response: [mathematical expression]{pm.response.status}`);',
' console.log(`Response time: {pm.response.responseTime}ms`);',
' ',
' } catch (e) {',
' console.log("Could not parse response as JSON");',
' }',
'}'
]
}
});
}
// Recursively process nested items
if (item.item) {
this.addScriptsToItems(item.item);
}
});
}
}
// Usage
const generator = new PostmanCollectionGenerator(
'./openapi.yaml',
'./postman-collection.json'
);
generator.generateCollection()
.then(() => console.log('Collection generated successfully'))
.catch(err => console.error('Generation failed:', err));
Testing and Quality Tools
API Testing Framework
# Comprehensive API testing framework
import pytest
import requests
import json
import time
from typing import Dict, Any, List
from dataclasses import dataclass
from urllib.parse import urljoin
@dataclass
class APITestResult:
endpoint: str
method: str
status_code: int
response_time: float
success: bool
error_message: str = None
class APITestSuite:
def __init__(self, base_url: str, api_key: str = None):
self.base_url = base_url
self.session = requests.Session()
if api_key:
self.session.headers.update({
'X-API-Key': api_key,
'Content-Type': 'application/json'
})
def test_endpoint(
self,
endpoint: str,
method: str = 'GET',
data: Dict = None,
expected_status: int = 200,
timeout: int = 30
) -> APITestResult:
"""Test a single API endpoint."""
url = urljoin(self.base_url, endpoint)
start_time = time.time()
try:
response = self.session.request(
method=method,
url=url,
json=data,
timeout=timeout
)
response_time = (time.time() - start_time) * 1000 # Convert to ms
success = response.status_code == expected_status
error_message = None if success else f"Expected {expected_status}, got {response.status_code}"
return APITestResult(
endpoint=endpoint,
method=method,
status_code=response.status_code,
response_time=response_time,
success=success,
error_message=error_message
)
except Exception as e:
return APITestResult(
endpoint=endpoint,
method=method,
status_code=0,
response_time=(time.time() - start_time) * 1000,
success=False,
error_message=str(e)
)
def run_smoke_tests(self) -> List[APITestResult]:
"""Run basic smoke tests on critical endpoints."""
smoke_tests = [
{'endpoint': '/health', 'method': 'GET', 'expected_status': 200},
{'endpoint': '/version', 'method': 'GET', 'expected_status': 200},
{'endpoint': '/users', 'method': 'GET', 'expected_status': 200},
{'endpoint': '/users/nonexistent', 'method': 'GET', 'expected_status': 404},
]
results = []
for test in smoke_tests:
result = self.test_endpoint(**test)
results.append(result)
return results
def run_load_tests(self, endpoint: str, concurrent_users: int = 10, duration: int = 60) -> Dict:
"""Run basic load testing on an endpoint."""
import threading
import queue
results_queue = queue.Queue()
def make_requests():
"""Worker function for load testing."""
start_time = time.time()
request_count = 0
while time.time() - start_time < duration:
result = self.test_endpoint(endpoint)
results_queue.put(result)
request_count += 1
time.sleep(0.1) # Small delay between requests
return request_count
# Start worker threads
threads = []
for _ in range(concurrent_users):
thread = threading.Thread(target=make_requests)
thread.start()
threads.append(thread)
# Wait for all threads to complete
for thread in threads:
thread.join()
# Collect results
all_results = []
while not results_queue.empty():
all_results.append(results_queue.get())
# Calculate statistics
successful_requests = [r for r in all_results if r.success]
failed_requests = [r for r in all_results if not r.success]
response_times = [r.response_time for r in successful_requests]
if response_times:
avg_response_time = sum(response_times) / len(response_times)
min_response_time = min(response_times)
max_response_time = max(response_times)
p95_response_time = sorted(response_times)[int(len(response_times) * 0.95)]
else:
avg_response_time = min_response_time = max_response_time = p95_response_time = 0
return {
'total_requests': len(all_results),
'successful_requests': len(successful_requests),
'failed_requests': len(failed_requests),
'success_rate': len(successful_requests) / len(all_results) * 100 if all_results else 0,
'avg_response_time': avg_response_time,
'min_response_time': min_response_time,
'max_response_time': max_response_time,
'p95_response_time': p95_response_time,
'requests_per_second': len(all_results) / duration
}
# pytest integration
class TestAPI:
@pytest.fixture(scope="session")
def api_client(self):
return APITestSuite(
base_url="https://api.example.com/v2",
api_key="test-api-key"
)
def test_api_health(self, api_client):
"""Test API health endpoint."""
result = api_client.test_endpoint('/health')
assert result.success, result.error_message
assert result.response_time < 1000, f"Health check too slow: {result.response_time}ms"
def test_user_operations(self, api_client):
"""Test user CRUD operations."""
# Create user
create_result = api_client.test_endpoint(
'/users',
method='POST',
data={
'name': 'Test User',
'email': 'test@example.com'
},
expected_status=201
)
assert create_result.success, create_result.error_message
# List users
list_result = api_client.test_endpoint('/users')
assert list_result.success, list_result.error_message
@pytest.mark.load
def test_api_load(self, api_client):
"""Load test critical endpoints."""
load_results = api_client.run_load_tests(
endpoint='/users',
concurrent_users=5,
duration=30
)
assert load_results['success_rate'] > 95, f"Success rate too low: {load_results['success_rate']}%"
assert load_results['p95_response_time'] < 2000, f"95th percentile too slow: {load_results['p95_response_time']}ms"
# Run tests
if __name__ == "__main__":
# Run smoke tests
api_client = APITestSuite("https://api.example.com/v2", "your-api-key")
print("Running smoke tests...")
smoke_results = api_client.run_smoke_tests()
for result in smoke_results:
status = "✅ PASS" if result.success else "❌ FAIL"
print(f"{status} {result.method} {result.endpoint} - {result.response_time:.0f}ms")
if not result.success:
print(f" Error: {result.error_message}")
print("\nRunning load test...")
load_results = api_client.run_load_tests('/users', concurrent_users=5, duration=30)
print(f"Load test results:")
print(f" Total requests: {load_results['total_requests']}")
print(f" Success rate: {load_results['success_rate']:.1f}%")
print(f" Avg response time: {load_results['avg_response_time']:.0f}ms")
print(f" 95th percentile: {load_results['p95_response_time']:.0f}ms")
print(f" Requests/second: {load_results['requests_per_second']:.1f}")
API management tools provide the essential infrastructure for building, deploying, and maintaining robust APIs. By leveraging these tools effectively, teams can focus on delivering business value while ensuring their APIs are secure, performant, and well-documented.