API Development for DevOps

1from fastapi import FastAPI, HTTPException, Depends, status
2from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3from pydantic import BaseModel, Field, validator
4from typing import List, Optional
5from datetime import datetime
6from enum import Enum
7import uvicorn
8
9app = FastAPI(
10    title="DevOps Automation API",
11    description="Production DevOps automation and orchestration",
12    version="1.0.0",
13    docs_url="/docs",
14    redoc_url="/redoc"
15)
16
17security = HTTPBearer()
18
19
20class DeploymentStatus(str, Enum):
21    """Deployment status enum"""
22    PENDING = "pending"
23    IN_PROGRESS = "in_progress"
24    COMPLETED = "completed"
25    FAILED = "failed"
26    ROLLED_BACK = "rolled_back"
27
28
29class DeploymentRequest(BaseModel):
30    """Deployment request model"""
31    service_name: str = Field(..., min_length=3, max_length=50)
32    version: str = Field(..., pattern=r"^\d+\.\d+\.\d+$")
33    environment: str = Field(..., pattern="^(dev|staging|production)$")
34    replicas: int = Field(default=3, ge=1, le=100)
35    rollback_on_failure: bool = Field(default=True)
36
37    @validator('service_name')
38    def validate_service_name(cls, v):
39        """Validate service name format"""
40        if not v.replace('-', '').replace('_', '').isalnum():
41            raise ValueError('Service name must be alphanumeric with hyphens/underscores')
42        return v
43
44    class Config:
45        schema_extra = {
46            "example": {
47                "service_name": "api-gateway",
48                "version": "1.2.3",
49                "environment": "production",
50                "replicas": 5,
51                "rollback_on_failure": True
52            }
53        }
54
55
56class DeploymentResponse(BaseModel):
57    """Deployment response model"""
58    deployment_id: str
59    status: DeploymentStatus
60    service_name: str
61    version: str
62    environment: str
63    created_at: datetime
64    message: str
65
66
67class InfrastructureRequest(BaseModel):
68    """Infrastructure provisioning request"""
69    resource_type: str = Field(..., pattern="^(ec2|rds|elasticache|s3)$")
70    name: str
71    instance_type: Optional[str] = "t3.medium"
72    region: str = "us-east-1"
73    tags: dict = {}
74
75
76# In-memory storage (use database in production)
77deployments_db = {}
78deployment_counter = 0
79
80
81@app.get("/", tags=["Health"])
82async def root():
83    """Root endpoint"""
84    return {
85        "service": "DevOps Automation API",
86        "version": "1.0.0",
87        "status": "operational"
88    }
89
90
91@app.get("/health", tags=["Health"])
92async def health_check():
93    """Health check endpoint"""
94    return {
95        "status": "healthy",
96        "timestamp": datetime.now().isoformat(),
97        "checks": {
98            "api": "ok",
99            "database": "ok",
100            "cache": "ok"
101        }
102    }
103
104
105@app.post(
106    "/api/v1/deployments",
107    response_model=DeploymentResponse,
108    status_code=status.HTTP_201_CREATED,
109    tags=["Deployments"]
110)
111async def create_deployment(
112    deployment: DeploymentRequest,
113    credentials: HTTPAuthorizationCredentials = Depends(security)
114):
115    """
116    Create a new deployment
117
118    This endpoint triggers a deployment workflow for the specified service.
119    The deployment will be executed asynchronously and status can be tracked
120    using the deployment_id.
121    """
122    global deployment_counter
123
124    # Validate token (simplified - use proper auth in production)
125    if not credentials.credentials.startswith("devops-"):
126        raise HTTPException(
127            status_code=status.HTTP_401_UNAUTHORIZED,
128            detail="Invalid authentication token"
129        )
130
131    # Create deployment
132    deployment_counter += 1
133    deployment_id = f"deploy-{deployment_counter:06d}"
134
135    deployment_record = DeploymentResponse(
136        deployment_id=deployment_id,
137        status=DeploymentStatus.PENDING,
138        service_name=deployment.service_name,
139        version=deployment.version,
140        environment=deployment.environment,
141        created_at=datetime.now(),
142        message=f"Deployment {deployment_id} queued successfully"
143    )
144
145    deployments_db[deployment_id] = deployment_record
146
147    # In production: Queue deployment task to background worker
148    # await deploy_service(deployment)
149
150    return deployment_record
151
152
153@app.get(
154    "/api/v1/deployments/{deployment_id}",
155    response_model=DeploymentResponse,
156    tags=["Deployments"]
157)
158async def get_deployment(
159    deployment_id: str,
160    credentials: HTTPAuthorizationCredentials = Depends(security)
161):
162    """Get deployment status by ID"""
163    if deployment_id not in deployments_db:
164        raise HTTPException(
165            status_code=status.HTTP_404_NOT_FOUND,
166            detail=f"Deployment {deployment_id} not found"
167        )
168
169    return deployments_db[deployment_id]
170
171
172@app.get(
173    "/api/v1/deployments",
174    response_model=List[DeploymentResponse],
175    tags=["Deployments"]
176)
177async def list_deployments(
178    environment: Optional[str] = None,
179    status: Optional[DeploymentStatus] = None,
180    limit: int = Field(default=10, le=100),
181    credentials: HTTPAuthorizationCredentials = Depends(security)
182):
183    """List deployments with optional filtering"""
184    results = list(deployments_db.values())
185
186    # Filter by environment
187    if environment:
188        results = [d for d in results if d.environment == environment]
189
190    # Filter by status
191    if status:
192        results = [d for d in results if d.status == status]
193
194    # Apply limit
195    results = results[:limit]
196
197    return results
198
199
200@app.post(
201    "/api/v1/infrastructure",
202    status_code=status.HTTP_202_ACCEPTED,
203    tags=["Infrastructure"]
204)
205async def provision_infrastructure(
206    infra: InfrastructureRequest,
207    credentials: HTTPAuthorizationCredentials = Depends(security)
208):
209    """
210    Provision infrastructure resources
211
212    Provisions AWS infrastructure asynchronously. Returns a task ID
213    that can be used to track provisioning status.
214    """
215    task_id = f"infra-{datetime.now().strftime('%Y%m%d%H%M%S')}"
216
217    # In production: Queue to Celery/RQ for async processing
218    # task = provision_aws_resource.delay(infra.dict())
219
220    return {
221        "task_id": task_id,
222        "status": "queued",
223        "resource_type": infra.resource_type,
224        "message": f"Infrastructure provisioning queued: {task_id}"
225    }
226
227
228if __name__ == "__main__":
229    # Run with: uvicorn main:app --reload
230    uvicorn.run(
231        "main:app",
232        host="0.0.0.0",
233        port=8000,
234        reload=True,
235        log_level="info"
236    )

1from fastapi import FastAPI, Depends, HTTPException, status
2from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
3from jose import JWTError, jwt
4from passlib.context import CryptContext
5from datetime import datetime, timedelta
6from typing import Optional
7from pydantic import BaseModel
8from enum import Enum
9
10# Configuration
11SECRET_KEY = "your-secret-key-change-in-production"
12ALGORITHM = "HS256"
13ACCESS_TOKEN_EXPIRE_MINUTES = 30
14
15pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
16oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
17
18app = FastAPI()
19
20
21class UserRole(str, Enum):
22    """User roles"""
23    ADMIN = "admin"
24    DEVELOPER = "developer"
25    VIEWER = "viewer"
26
27
28class User(BaseModel):
29    """User model"""
30    username: str
31    email: Optional[str] = None
32    role: UserRole = UserRole.VIEWER
33    disabled: bool = False
34
35
36class UserInDB(User):
37    """User with hashed password"""
38    hashed_password: str
39
40
41class Token(BaseModel):
42    """JWT token response"""
43    access_token: str
44    token_type: str
45
46
47class TokenData(BaseModel):
48    """Token payload data"""
49    username: Optional[str] = None
50    role: Optional[UserRole] = None
51
52
53# Mock user database
54fake_users_db = {
55    "admin": {
56        "username": "admin",
57        "email": "[email protected]",
58        "role": UserRole.ADMIN,
59        "hashed_password": pwd_context.hash("admin123"),
60        "disabled": False,
61    },
62    "developer": {
63        "username": "developer",
64        "email": "[email protected]",
65        "role": UserRole.DEVELOPER,
66        "hashed_password": pwd_context.hash("dev123"),
67        "disabled": False,
68    }
69}
70
71
72def verify_password(plain_password: str, hashed_password: str) -> bool:
73    """Verify password against hash"""
74    return pwd_context.verify(plain_password, hashed_password)
75
76
77def get_password_hash(password: str) -> str:
78    """Hash password"""
79    return pwd_context.hash(password)
80
81
82def get_user(username: str) -> Optional[UserInDB]:
83    """Get user from database"""
84    if username in fake_users_db:
85        user_dict = fake_users_db[username]
86        return UserInDB(**user_dict)
87
88
89def authenticate_user(username: str, password: str) -> Optional[UserInDB]:
90    """Authenticate user"""
91    user = get_user(username)
92    if not user:
93        return None
94    if not verify_password(password, user.hashed_password):
95        return None
96    return user
97
98
99def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
100    """Create JWT access token"""
101    to_encode = data.copy()
102    if expires_delta:
103        expire = datetime.utcnow() + expires_delta
104    else:
105        expire = datetime.utcnow() + timedelta(minutes=15)
106
107    to_encode.update({"exp": expire})
108    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
109    return encoded_jwt
110
111
112async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
113    """Get current authenticated user"""
114    credentials_exception = HTTPException(
115        status_code=status.HTTP_401_UNAUTHORIZED,
116        detail="Could not validate credentials",
117        headers={"WWW-Authenticate": "Bearer"},
118    )
119
120    try:
121        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
122        username: str = payload.get("sub")
123        role: str = payload.get("role")
124
125        if username is None:
126            raise credentials_exception
127
128        token_data = TokenData(username=username, role=role)
129    except JWTError:
130        raise credentials_exception
131
132    user = get_user(username=token_data.username)
133    if user is None:
134        raise credentials_exception
135
136    return user
137
138
139async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
140    """Get current active user"""
141    if current_user.disabled:
142        raise HTTPException(status_code=400, detail="Inactive user")
143    return current_user
144
145
146def require_role(required_role: UserRole):
147    """Dependency to require specific role"""
148    async def role_checker(current_user: User = Depends(get_current_active_user)):
149        role_hierarchy = {
150            UserRole.ADMIN: 3,
151            UserRole.DEVELOPER: 2,
152            UserRole.VIEWER: 1
153        }
154
155        if role_hierarchy.get(current_user.role, 0) < role_hierarchy.get(required_role, 0):
156            raise HTTPException(
157                status_code=status.HTTP_403_FORBIDDEN,
158                detail="Insufficient permissions"
159            )
160        return current_user
161
162    return role_checker
163
164
165@app.post("/token", response_model=Token)
166async def login(form_data: OAuth2PasswordRequestForm = Depends()):
167    """Login endpoint - returns JWT token"""
168    user = authenticate_user(form_data.username, form_data.password)
169
170    if not user:
171        raise HTTPException(
172            status_code=status.HTTP_401_UNAUTHORIZED,
173            detail="Incorrect username or password",
174            headers={"WWW-Authenticate": "Bearer"},
175        )
176
177    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
178    access_token = create_access_token(
179        data={"sub": user.username, "role": user.role.value},
180        expires_delta=access_token_expires
181    )
182
183    return {"access_token": access_token, "token_type": "bearer"}
184
185
186@app.get("/users/me", response_model=User)
187async def read_users_me(current_user: User = Depends(get_current_active_user)):
188    """Get current user info"""
189    return current_user
190
191
192@app.post("/api/v1/deployments", dependencies=[Depends(require_role(UserRole.DEVELOPER))])
193async def create_deployment_protected(deployment: dict):
194    """Create deployment - requires developer role or higher"""
195    return {"status": "deployment created", "deployment": deployment}
196
197
198@app.delete("/api/v1/infrastructure/{resource_id}", dependencies=[Depends(require_role(UserRole.ADMIN))])
199async def delete_infrastructure_protected(resource_id: str):
200    """Delete infrastructure - requires admin role"""
201    return {"status": "resource deleted", "resource_id": resource_id}

1from fastapi import FastAPI, Request
2from slowapi import Limiter, _rate_limit_exceeded_handler
3from slowapi.util import get_remote_address
4from slowapi.errors import RateLimitExceeded
5
6limiter = Limiter(key_func=get_remote_address)
7app = FastAPI()
8app.state.limiter = limiter
9app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
10
11
12@app.get("/api/v1/deployments")
13@limiter.limit("100/minute")  # 100 requests per minute
14async def list_deployments(request: Request):
15    """List deployments with rate limiting"""
16    return {"deployments": []}
17
18
19@app.post("/api/v1/infrastructure")
20@limiter.limit("10/minute")  # More restrictive for expensive operations
21async def provision_infrastructure(request: Request, infra: dict):
22    """Provision infrastructure with strict rate limiting"""
23    return {"status": "queued"}
24
25
26# Custom rate limit by user
27@app.get("/api/v1/metrics")
28@limiter.limit("1000/hour", key_func=lambda request: request.headers.get("X-User-ID", "anonymous"))
29async def get_metrics(request: Request):
30    """Get metrics with per-user rate limiting"""
31    return {"metrics": {}}

1from fastapi import FastAPI, Request
2from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
3from starlette.responses import Response
4import time
5
6app = FastAPI()
7
8# Define metrics
9http_requests_total = Counter(
10    'http_requests_total',
11    'Total HTTP requests',
12    ['method', 'endpoint', 'status']
13)
14
15http_request_duration_seconds = Histogram(
16    'http_request_duration_seconds',
17    'HTTP request duration',
18    ['method', 'endpoint']
19)
20
21active_requests = Gauge(
22    'active_requests',
23    'Number of active requests'
24)
25
26deployment_count = Counter(
27    'deployment_count_total',
28    'Total number of deployments',
29    ['environment', 'status']
30)
31
32
33@app.middleware("http")
34async def prometheus_middleware(request: Request, call_next):
35    """Middleware to track request metrics"""
36    active_requests.inc()
37    start_time = time.time()
38
39    response = await call_next(request)
40
41    duration = time.time() - start_time
42
43    http_requests_total.labels(
44        method=request.method,
45        endpoint=request.url.path,
46        status=response.status_code
47    ).inc()
48
49    http_request_duration_seconds.labels(
50        method=request.method,
51        endpoint=request.url.path
52    ).observe(duration)
53
54    active_requests.dec()
55
56    return response
57
58
59@app.get("/metrics")
60async def metrics():
61    """Prometheus metrics endpoint"""
62    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)
63
64
65@app.post("/api/v1/deployments")
66async def create_deployment(deployment: dict):
67    """Create deployment with metrics tracking"""
68    # Track deployment
69    deployment_count.labels(
70        environment=deployment.get('environment', 'unknown'),
71        status='initiated'
72    ).inc()
73
74    # Deployment logic...
75
76    return {"status": "created"}

1import pytest
2from fastapi.testclient import TestClient
3from main import app
4
5client = TestClient(app)
6
7
8def test_health_check():
9    """Test health check endpoint"""
10    response = client.get("/health")
11    assert response.status_code == 200
12    assert response.json()["status"] == "healthy"
13
14
15def test_create_deployment_unauthorized():
16    """Test deployment without auth fails"""
17    response = client.post("/api/v1/deployments", json={
18        "service_name": "test-service",
19        "version": "1.0.0",
20        "environment": "dev"
21    })
22    assert response.status_code == 403
23
24
25def test_create_deployment_authorized():
26    """Test deployment with valid auth succeeds"""
27    # First, get token
28    login_response = client.post("/token", data={
29        "username": "developer",
30        "password": "dev123"
31    })
32    token = login_response.json()["access_token"]
33
34    # Create deployment
35    response = client.post(
36        "/api/v1/deployments",
37        json={
38            "service_name": "test-service",
39            "version": "1.0.0",
40            "environment": "dev",
41            "replicas": 3
42        },
43        headers={"Authorization": f"Bearer {token}"}
44    )
45
46    assert response.status_code == 201
47    assert "deployment_id" in response.json()
48
49
50def test_rate_limiting():
51    """Test rate limiting works"""
52    # Make requests until rate limited
53    for _ in range(101):
54        response = client.get("/api/v1/deployments")
55
56    assert response.status_code == 429  # Too Many Requests
57
58
59def test_input_validation():
60    """Test pydantic validation"""
61    response = client.post("/api/v1/deployments", json={
62        "service_name": "invalid service!",  # Invalid characters
63        "version": "abc",  # Invalid version format
64        "environment": "invalid"  # Invalid environment
65    })
66
67    assert response.status_code == 422
68    assert "detail" in response.json()
69
70
71# Run tests with: pytest -v

1# Dockerfile for FastAPI production
2FROM python:3.11-slim
3
4WORKDIR /app
5
6# Install dependencies
7COPY requirements.txt .
8RUN pip install --no-cache-dir -r requirements.txt
9
10# Copy application
11COPY . .
12
13# Create non-root user
14RUN useradd -m -u 1000 apiuser && chown -R apiuser:apiuser /app
15USER apiuser
16
17# Run with gunicorn for production
18CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]