Platform APIs and Abstractions

📖 Introduction

One of the most important lessons I've learned in platform engineering is that the API is the product. No matter how sophisticated your underlying infrastructure, if developers can't interact with it through well-designed APIs, adoption will struggle.

Platform APIs create the abstraction layer that hides infrastructure complexity while exposing the capabilities developers actually need. In this article, we'll explore how to design platform APIs, leverage Kubernetes operators for automation, and use tools like Crossplane for infrastructure abstraction.

---

🎯 What Are Platform APIs?

Definition

Platform APIs are programmatic interfaces that expose platform capabilities as self-service resources. They abstract underlying infrastructure complexity while maintaining the flexibility developers need.

API Design Principles

Principle	Description	Example
Declarative	Describe desired state, not steps	"I want a database" not "create RDS, configure VPC..."
Self-Describing	APIs document themselves	OpenAPI specs, CRD schemas
Versioned	Backward compatible evolution	v1alpha1 → v1beta1 → v1
Consistent	Predictable patterns	Same structure across resources
Observable	Status reflects reality	Conditions, events, metrics

---

🔧 Kubernetes Custom Resources

The Power of CRDs

Custom Resource Definitions (CRDs) extend Kubernetes with your own resource types, creating a declarative API for your platform.

# Example: Custom Application Resource
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: applications.platform.example.com
spec:
  group: platform.example.com
  names:
    kind: Application
    listKind: ApplicationList
    plural: applications
    singular: application
    shortNames:
      - app
  scope: Namespaced
  versions:
    - name: v1alpha1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              required:
                - team
                - image
              properties:
                team:
                  type: string
                  description: Team that owns this application
                image:
                  type: string
                  description: Container image to deploy
                replicas:
                  type: integer
                  minimum: 1
                  maximum: 20
                  default: 3
                tier:
                  type: string
                  enum: [development, staging, production]
                  default: development
                resources:
                  type: object
                  properties:
                    cpu:
                      type: string
                      default: "500m"
                    memory:
                      type: string
                      default: "512Mi"
                database:
                  type: object
                  properties:
                    enabled:
                      type: boolean
                      default: false
                    type:
                      type: string
                      enum: [postgres, mysql]
                    size:
                      type: string
                      enum: [small, medium, large]
            status:
              type: object
              properties:
                phase:
                  type: string
                conditions:
                  type: array
                  items:
                    type: object
                    properties:
                      type:
                        type: string
                      status:
                        type: string
                      message:
                        type: string
                      lastTransitionTime:
                        type: string
                        format: date-time
      subresources:
        status: {}
      additionalPrinterColumns:
        - name: Team
          type: string
          jsonPath: .spec.team
        - name: Tier
          type: string
          jsonPath: .spec.tier
        - name: Phase
          type: string
          jsonPath: .status.phase
        - name: Age
          type: date
          jsonPath: .metadata.creationTimestamp

Using the Custom Resource

# Developers create applications like this
apiVersion: platform.example.com/v1alpha1
kind: Application
metadata:
  name: order-service
  namespace: checkout-team
spec:
  team: checkout
  image: registry.example.com/checkout/order-service:v1.2.3
  replicas: 3
  tier: production
  resources:
    cpu: "1"
    memory: "1Gi"
  database:
    enabled: true
    type: postgres
    size: medium

---

🎮 Kubernetes Operators

Operator Pattern

Operators encode operational knowledge into software, automating the lifecycle management of applications and infrastructure.

Python Operator with Kopf

"""
Platform Application Operator using Kopf.
Manages the lifecycle of Application custom resources.
"""

import kopf
import kubernetes
from kubernetes import client
from typing import Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
import logging

logger = logging.getLogger(__name__)


class ApplicationPhase(Enum):
    PENDING = "Pending"
    PROVISIONING = "Provisioning"
    RUNNING = "Running"
    FAILED = "Failed"
    TERMINATING = "Terminating"


@dataclass
class ApplicationSpec:
    """Parsed application specification."""
    team: str
    image: str
    replicas: int = 3
    tier: str = "development"
    cpu: str = "500m"
    memory: str = "512Mi"
    database_enabled: bool = False
    database_type: Optional[str] = None
    database_size: Optional[str] = None
    
    @classmethod
    def from_dict(cls, spec: Dict[str, Any]) -> 'ApplicationSpec':
        resources = spec.get('resources', {})
        database = spec.get('database', {})
        
        return cls(
            team=spec['team'],
            image=spec['image'],
            replicas=spec.get('replicas', 3),
            tier=spec.get('tier', 'development'),
            cpu=resources.get('cpu', '500m'),
            memory=resources.get('memory', '512Mi'),
            database_enabled=database.get('enabled', False),
            database_type=database.get('type'),
            database_size=database.get('size'),
        )


class ApplicationReconciler:
    """Reconciles Application resources to desired state."""
    
    def __init__(self):
        kubernetes.config.load_incluster_config()
        self.apps_api = client.AppsV1Api()
        self.core_api = client.CoreV1Api()
        self.custom_api = client.CustomObjectsApi()
    
    def reconcile(self, name: str, namespace: str, spec: ApplicationSpec) -> Dict[str, Any]:
        """Reconcile application to desired state."""
        try:
            # Create or update Deployment
            self._reconcile_deployment(name, namespace, spec)
            
            # Create or update Service
            self._reconcile_service(name, namespace, spec)
            
            # Create database if needed
            if spec.database_enabled:
                self._reconcile_database(name, namespace, spec)
            
            return {
                "phase": ApplicationPhase.RUNNING.value,
                "conditions": [
                    {
                        "type": "Ready",
                        "status": "True",
                        "message": "Application is running",
                    }
                ],
            }
        
        except Exception as e:
            logger.error(f"Reconciliation failed: {e}")
            return {
                "phase": ApplicationPhase.FAILED.value,
                "conditions": [
                    {
                        "type": "Ready",
                        "status": "False",
                        "message": str(e),
                    }
                ],
            }
    
    def _reconcile_deployment(self, name: str, namespace: str, spec: ApplicationSpec):
        """Create or update Deployment."""
        deployment = client.V1Deployment(
            api_version="apps/v1",
            kind="Deployment",
            metadata=client.V1ObjectMeta(
                name=name,
                namespace=namespace,
                labels={
                    "app": name,
                    "team": spec.team,
                    "tier": spec.tier,
                    "managed-by": "platform-operator",
                },
            ),
            spec=client.V1DeploymentSpec(
                replicas=spec.replicas,
                selector=client.V1LabelSelector(
                    match_labels={"app": name}
                ),
                template=client.V1PodTemplateSpec(
                    metadata=client.V1ObjectMeta(
                        labels={
                            "app": name,
                            "team": spec.team,
                        },
                        annotations={
                            "prometheus.io/scrape": "true",
                            "prometheus.io/port": "8080",
                        },
                    ),
                    spec=client.V1PodSpec(
                        containers=[
                            client.V1Container(
                                name=name,
                                image=spec.image,
                                ports=[
                                    client.V1ContainerPort(container_port=8080)
                                ],
                                resources=client.V1ResourceRequirements(
                                    requests={
                                        "cpu": spec.cpu,
                                        "memory": spec.memory,
                                    },
                                    limits={
                                        "cpu": spec.cpu,
                                        "memory": spec.memory,
                                    },
                                ),
                                liveness_probe=client.V1Probe(
                                    http_get=client.V1HTTPGetAction(
                                        path="/health",
                                        port=8080,
                                    ),
                                    initial_delay_seconds=10,
                                    period_seconds=10,
                                ),
                                readiness_probe=client.V1Probe(
                                    http_get=client.V1HTTPGetAction(
                                        path="/ready",
                                        port=8080,
                                    ),
                                    initial_delay_seconds=5,
                                    period_seconds=5,
                                ),
                                env=self._get_environment_variables(name, spec),
                            )
                        ],
                        security_context=client.V1PodSecurityContext(
                            run_as_non_root=True,
                            run_as_user=1000,
                        ),
                    ),
                ),
            ),
        )
        
        try:
            self.apps_api.read_namespaced_deployment(name, namespace)
            self.apps_api.patch_namespaced_deployment(name, namespace, deployment)
            logger.info(f"Updated deployment {name}")
        except kubernetes.client.exceptions.ApiException as e:
            if e.status == 404:
                self.apps_api.create_namespaced_deployment(namespace, deployment)
                logger.info(f"Created deployment {name}")
            else:
                raise
    
    def _reconcile_service(self, name: str, namespace: str, spec: ApplicationSpec):
        """Create or update Service."""
        service = client.V1Service(
            api_version="v1",
            kind="Service",
            metadata=client.V1ObjectMeta(
                name=name,
                namespace=namespace,
                labels={
                    "app": name,
                    "team": spec.team,
                },
            ),
            spec=client.V1ServiceSpec(
                selector={"app": name},
                ports=[
                    client.V1ServicePort(
                        port=80,
                        target_port=8080,
                    )
                ],
            ),
        )
        
        try:
            self.core_api.read_namespaced_service(name, namespace)
            self.core_api.patch_namespaced_service(name, namespace, service)
        except kubernetes.client.exceptions.ApiException as e:
            if e.status == 404:
                self.core_api.create_namespaced_service(namespace, service)
            else:
                raise
    
    def _reconcile_database(self, name: str, namespace: str, spec: ApplicationSpec):
        """Create database resource using Crossplane."""
        database_claim = {
            "apiVersion": "database.platform.example.com/v1alpha1",
            "kind": "PostgreSQLClaim",
            "metadata": {
                "name": f"{name}-db",
                "namespace": namespace,
            },
            "spec": {
                "size": spec.database_size,
                "team": spec.team,
            },
        }
        
        try:
            self.custom_api.get_namespaced_custom_object(
                group="database.platform.example.com",
                version="v1alpha1",
                namespace=namespace,
                plural="postgresqlclaims",
                name=f"{name}-db",
            )
            # Update if exists
            self.custom_api.patch_namespaced_custom_object(
                group="database.platform.example.com",
                version="v1alpha1",
                namespace=namespace,
                plural="postgresqlclaims",
                name=f"{name}-db",
                body=database_claim,
            )
        except kubernetes.client.exceptions.ApiException as e:
            if e.status == 404:
                self.custom_api.create_namespaced_custom_object(
                    group="database.platform.example.com",
                    version="v1alpha1",
                    namespace=namespace,
                    plural="postgresqlclaims",
                    body=database_claim,
                )
            else:
                raise
    
    def _get_environment_variables(self, name: str, spec: ApplicationSpec) -> list:
        """Generate environment variables for the container."""
        env_vars = [
            client.V1EnvVar(name="SERVICE_NAME", value=name),
            client.V1EnvVar(name="TEAM", value=spec.team),
            client.V1EnvVar(name="TIER", value=spec.tier),
        ]
        
        if spec.database_enabled:
            env_vars.extend([
                client.V1EnvVar(
                    name="DATABASE_URL",
                    value_from=client.V1EnvVarSource(
                        secret_key_ref=client.V1SecretKeySelector(
                            name=f"{name}-db-credentials",
                            key="url",
                        )
                    ),
                ),
            ])
        
        return env_vars


# Kopf handlers
reconciler = ApplicationReconciler()


@kopf.on.create('platform.example.com', 'v1alpha1', 'applications')
def on_create(spec, name, namespace, **kwargs):
    """Handle Application creation."""
    logger.info(f"Creating application {name} in {namespace}")
    app_spec = ApplicationSpec.from_dict(spec)
    status = reconciler.reconcile(name, namespace, app_spec)
    return status


@kopf.on.update('platform.example.com', 'v1alpha1', 'applications')
def on_update(spec, name, namespace, **kwargs):
    """Handle Application updates."""
    logger.info(f"Updating application {name} in {namespace}")
    app_spec = ApplicationSpec.from_dict(spec)
    status = reconciler.reconcile(name, namespace, app_spec)
    return status


@kopf.on.delete('platform.example.com', 'v1alpha1', 'applications')
def on_delete(spec, name, namespace, **kwargs):
    """Handle Application deletion."""
    logger.info(f"Deleting application {name} from {namespace}")
    # Kubernetes garbage collection handles child resources
    # if owner references are set correctly

---

🌉 Crossplane for Infrastructure

What is Crossplane?

Crossplane extends Kubernetes to manage any infrastructure resource using the same declarative approach as Kubernetes resources.

Crossplane Composition

# CompositeResourceDefinition - Define the platform abstraction
apiVersion: apiextensions.crossplane.io/v1
kind: CompositeResourceDefinition
metadata:
  name: xdatabases.platform.example.com
spec:
  group: platform.example.com
  names:
    kind: XDatabase
    plural: xdatabases
  claimNames:
    kind: DatabaseClaim
    plural: databaseclaims
  versions:
    - name: v1alpha1
      served: true
      referenceable: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              required:
                - size
              properties:
                size:
                  type: string
                  enum: [small, medium, large]
                  description: Database size tier
                engine:
                  type: string
                  enum: [postgres, mysql]
                  default: postgres
                version:
                  type: string
                  default: "15"
            status:
              type: object
              properties:
                endpoint:
                  type: string
                port:
                  type: integer
                ready:
                  type: boolean

---
# Composition - How to fulfill the abstraction
apiVersion: apiextensions.crossplane.io/v1
kind: Composition
metadata:
  name: database-aws
  labels:
    provider: aws
    engine: postgres
spec:
  compositeTypeRef:
    apiVersion: platform.example.com/v1alpha1
    kind: XDatabase
  
  resources:
    # RDS Instance
    - name: rds-instance
      base:
        apiVersion: rds.aws.crossplane.io/v1beta1
        kind: Instance
        spec:
          forProvider:
            engine: postgres
            engineVersion: "15"
            instanceClass: db.t3.micro
            allocatedStorage: 20
            masterUsername: admin
            skipFinalSnapshot: true
            publiclyAccessible: false
            vpcSecurityGroupIDRefs:
              - name: db-security-group
            dbSubnetGroupNameRef:
              name: db-subnet-group
          writeConnectionSecretToRef:
            namespace: crossplane-system
      patches:
        # Map size to instance class
        - type: FromCompositeFieldPath
          fromFieldPath: spec.size
          toFieldPath: spec.forProvider.instanceClass
          transforms:
            - type: map
              map:
                small: db.t3.micro
                medium: db.t3.medium
                large: db.r5.large
        
        # Map size to storage
        - type: FromCompositeFieldPath
          fromFieldPath: spec.size
          toFieldPath: spec.forProvider.allocatedStorage
          transforms:
            - type: map
              map:
                small: 20
                medium: 100
                large: 500
        
        # Pass endpoint to status
        - type: ToCompositeFieldPath
          fromFieldPath: status.atProvider.endpoint.address
          toFieldPath: status.endpoint
        
        - type: ToCompositeFieldPath
          fromFieldPath: status.atProvider.endpoint.port
          toFieldPath: status.port
      
      connectionDetails:
        - name: endpoint
          fromFieldPath: status.atProvider.endpoint.address
        - name: port
          fromFieldPath: status.atProvider.endpoint.port
        - name: username
          fromConnectionSecretKey: username
        - name: password
          fromConnectionSecretKey: password

    # Security Group
    - name: security-group
      base:
        apiVersion: ec2.aws.crossplane.io/v1beta1
        kind: SecurityGroup
        metadata:
          name: db-security-group
        spec:
          forProvider:
            region: us-east-1
            vpcIdRef:
              name: platform-vpc
            description: Security group for platform databases
            ingress:
              - fromPort: 5432
                toPort: 5432
                ipProtocol: tcp
                ipRanges:
                  - cidrIp: 10.0.0.0/8

---
# How developers use it (DatabaseClaim)
apiVersion: platform.example.com/v1alpha1
kind: DatabaseClaim
metadata:
  name: orders-db
  namespace: checkout-team
spec:
  size: medium
  engine: postgres
  
  # Connection secret will be created here
  writeConnectionSecretToRef:
    name: orders-db-credentials

Python Client for Crossplane Resources

"""
Client for managing Crossplane resources.
"""

from dataclasses import dataclass
from typing import Dict, Any, Optional, List
from kubernetes import client, config
import time


@dataclass
class DatabaseStatus:
    """Status of a database claim."""
    ready: bool
    endpoint: Optional[str]
    port: Optional[int]
    message: Optional[str]


class CrossplaneClient:
    """Client for interacting with Crossplane resources."""
    
    def __init__(self):
        config.load_incluster_config()
        self.custom_api = client.CustomObjectsApi()
    
    def create_database(
        self,
        name: str,
        namespace: str,
        size: str = "small",
        engine: str = "postgres",
    ) -> Dict[str, Any]:
        """Create a database claim."""
        claim = {
            "apiVersion": "platform.example.com/v1alpha1",
            "kind": "DatabaseClaim",
            "metadata": {
                "name": name,
                "namespace": namespace,
            },
            "spec": {
                "size": size,
                "engine": engine,
                "writeConnectionSecretToRef": {
                    "name": f"{name}-credentials",
                },
            },
        }
        
        return self.custom_api.create_namespaced_custom_object(
            group="platform.example.com",
            version="v1alpha1",
            namespace=namespace,
            plural="databaseclaims",
            body=claim,
        )
    
    def get_database_status(
        self, 
        name: str, 
        namespace: str
    ) -> DatabaseStatus:
        """Get the status of a database claim."""
        try:
            claim = self.custom_api.get_namespaced_custom_object(
                group="platform.example.com",
                version="v1alpha1",
                namespace=namespace,
                plural="databaseclaims",
                name=name,
            )
            
            status = claim.get("status", {})
            conditions = status.get("conditions", [])
            
            ready = any(
                c.get("type") == "Ready" and c.get("status") == "True"
                for c in conditions
            )
            
            return DatabaseStatus(
                ready=ready,
                endpoint=status.get("endpoint"),
                port=status.get("port"),
                message=conditions[-1].get("message") if conditions else None,
            )
        
        except client.exceptions.ApiException as e:
            if e.status == 404:
                return DatabaseStatus(
                    ready=False,
                    endpoint=None,
                    port=None,
                    message="Database not found",
                )
            raise
    
    def wait_for_database(
        self,
        name: str,
        namespace: str,
        timeout: int = 600,
    ) -> DatabaseStatus:
        """Wait for database to become ready."""
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            status = self.get_database_status(name, namespace)
            if status.ready:
                return status
            time.sleep(10)
        
        raise TimeoutError(f"Database {name} not ready after {timeout} seconds")
    
    def delete_database(self, name: str, namespace: str) -> bool:
        """Delete a database claim."""
        try:
            self.custom_api.delete_namespaced_custom_object(
                group="platform.example.com",
                version="v1alpha1",
                namespace=namespace,
                plural="databaseclaims",
                name=name,
            )
            return True
        except client.exceptions.ApiException as e:
            if e.status == 404:
                return True  # Already deleted
            raise
    
    def list_databases(self, namespace: str) -> List[Dict[str, Any]]:
        """List all database claims in a namespace."""
        response = self.custom_api.list_namespaced_custom_object(
            group="platform.example.com",
            version="v1alpha1",
            namespace=namespace,
            plural="databaseclaims",
        )
        return response.get("items", [])


# Usage
async def provision_service_with_database():
    """Example: Provision a service with its database."""
    crossplane = CrossplaneClient()
    
    # Create database
    crossplane.create_database(
        name="orders-db",
        namespace="checkout-team",
        size="medium",
        engine="postgres",
    )
    
    # Wait for database to be ready
    status = crossplane.wait_for_database(
        name="orders-db",
        namespace="checkout-team",
        timeout=600,
    )
    
    print(f"Database ready at {status.endpoint}:{status.port}")

---

🔌 REST API Design

Platform API Gateway

"""
Platform REST API using FastAPI.
Provides HTTP interface to platform capabilities.
"""

from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum
import uuid
from datetime import datetime

app = FastAPI(
    title="Platform API",
    description="Self-service platform capabilities",
    version="1.0.0",
)

security = HTTPBearer()


# Models
class ResourceSize(str, Enum):
    SMALL = "small"
    MEDIUM = "medium"
    LARGE = "large"


class ResourceTier(str, Enum):
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"


class ApplicationCreate(BaseModel):
    name: str = Field(..., regex=r"^[a-z][a-z0-9-]*$")
    team: str
    image: str
    replicas: int = Field(default=3, ge=1, le=20)
    tier: ResourceTier = ResourceTier.DEVELOPMENT
    size: ResourceSize = ResourceSize.SMALL
    database: Optional[str] = None  # postgres, mysql, or None


class ApplicationResponse(BaseModel):
    id: str
    name: str
    team: str
    tier: ResourceTier
    status: str
    created_at: datetime
    endpoint: Optional[str] = None


class DatabaseCreate(BaseModel):
    name: str = Field(..., regex=r"^[a-z][a-z0-9-]*$")
    engine: str = "postgres"
    size: ResourceSize = ResourceSize.SMALL


class DatabaseResponse(BaseModel):
    id: str
    name: str
    engine: str
    size: ResourceSize
    status: str
    endpoint: Optional[str] = None
    port: Optional[int] = None


class EnvironmentCreate(BaseModel):
    name: str = Field(..., regex=r"^[a-z][a-z0-9-]*$")
    tier: ResourceTier
    expires_in_days: Optional[int] = Field(default=7, ge=1, le=30)


class EnvironmentResponse(BaseModel):
    id: str
    name: str
    tier: ResourceTier
    status: str
    kubeconfig_secret: Optional[str] = None
    expires_at: Optional[datetime] = None


# Dependencies
async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    """Validate token and return user info."""
    # In production, validate JWT token
    return {
        "user_id": "user123",
        "team": "checkout-team",
        "roles": ["developer"],
    }


async def check_team_access(
    team: str, 
    user: dict = Depends(get_current_user)
) -> bool:
    """Verify user has access to the team's resources."""
    # In production, check actual permissions
    return user["team"] == team or "admin" in user["roles"]


# Routes
@app.post("/api/v1/applications", response_model=ApplicationResponse)
async def create_application(
    app_request: ApplicationCreate,
    background_tasks: BackgroundTasks,
    user: dict = Depends(get_current_user),
):
    """Create a new application."""
    if not await check_team_access(app_request.team, user):
        raise HTTPException(status_code=403, detail="Access denied to team")
    
    # Validate production tier requirements
    if app_request.tier == ResourceTier.PRODUCTION:
        if app_request.replicas < 3:
            raise HTTPException(
                status_code=400,
                detail="Production apps require at least 3 replicas"
            )
    
    app_id = str(uuid.uuid4())
    
    # Trigger async provisioning
    background_tasks.add_task(
        provision_application,
        app_id,
        app_request,
    )
    
    return ApplicationResponse(
        id=app_id,
        name=app_request.name,
        team=app_request.team,
        tier=app_request.tier,
        status="provisioning",
        created_at=datetime.utcnow(),
    )


@app.get("/api/v1/applications", response_model=List[ApplicationResponse])
async def list_applications(
    team: Optional[str] = None,
    tier: Optional[ResourceTier] = None,
    user: dict = Depends(get_current_user),
):
    """List applications."""
    # In production, query from database/Kubernetes
    return []


@app.get("/api/v1/applications/{app_id}", response_model=ApplicationResponse)
async def get_application(
    app_id: str,
    user: dict = Depends(get_current_user),
):
    """Get application details."""
    # In production, fetch from database/Kubernetes
    raise HTTPException(status_code=404, detail="Application not found")


@app.delete("/api/v1/applications/{app_id}")
async def delete_application(
    app_id: str,
    background_tasks: BackgroundTasks,
    user: dict = Depends(get_current_user),
):
    """Delete an application."""
    background_tasks.add_task(deprovision_application, app_id)
    return {"status": "deleting", "id": app_id}


@app.post("/api/v1/databases", response_model=DatabaseResponse)
async def create_database(
    db_request: DatabaseCreate,
    background_tasks: BackgroundTasks,
    user: dict = Depends(get_current_user),
):
    """Create a new database."""
    db_id = str(uuid.uuid4())
    
    background_tasks.add_task(
        provision_database,
        db_id,
        db_request,
        user["team"],
    )
    
    return DatabaseResponse(
        id=db_id,
        name=db_request.name,
        engine=db_request.engine,
        size=db_request.size,
        status="provisioning",
    )


@app.post("/api/v1/environments", response_model=EnvironmentResponse)
async def create_environment(
    env_request: EnvironmentCreate,
    background_tasks: BackgroundTasks,
    user: dict = Depends(get_current_user),
):
    """Create a new environment."""
    # Validate tier constraints
    if env_request.tier == ResourceTier.PRODUCTION:
        raise HTTPException(
            status_code=400,
            detail="Production environments require approval workflow"
        )
    
    env_id = str(uuid.uuid4())
    
    background_tasks.add_task(
        provision_environment,
        env_id,
        env_request,
        user["team"],
    )
    
    return EnvironmentResponse(
        id=env_id,
        name=env_request.name,
        tier=env_request.tier,
        status="provisioning",
        expires_at=datetime.utcnow() if env_request.expires_in_days else None,
    )


# Background tasks
async def provision_application(app_id: str, request: ApplicationCreate):
    """Provision application resources."""
    # Create Kubernetes Application custom resource
    pass


async def deprovision_application(app_id: str):
    """Remove application resources."""
    pass


async def provision_database(db_id: str, request: DatabaseCreate, team: str):
    """Provision database via Crossplane."""
    pass


async def provision_environment(env_id: str, request: EnvironmentCreate, team: str):
    """Provision environment namespace and resources."""
    pass


# Health endpoints
@app.get("/health")
async def health():
    return {"status": "healthy"}


@app.get("/ready")
async def ready():
    # Check dependencies
    return {"status": "ready"}

---

📊 API Versioning Strategy

Version Evolution

v1alpha1 → v1beta1 → v1 → v2alpha1 → v2
    ↓           ↓       ↓
 Breaking   Stable   Production
 changes    testing    ready
 expected

Conversion Webhook

"""
Conversion webhook for API version migration.
"""

from fastapi import FastAPI, Request
from pydantic import BaseModel
from typing import Dict, Any, List

app = FastAPI()


class ConversionRequest(BaseModel):
    apiVersion: str
    kind: str
    request: Dict[str, Any]


class ConversionResponse(BaseModel):
    apiVersion: str
    kind: str
    response: Dict[str, Any]


def convert_v1alpha1_to_v1beta1(obj: Dict[str, Any]) -> Dict[str, Any]:
    """Convert v1alpha1 to v1beta1."""
    # Example: Rename field
    spec = obj.get("spec", {})
    
    # In v1alpha1, we had "size", in v1beta1 we have "resourceProfile"
    if "size" in spec:
        spec["resourceProfile"] = spec.pop("size")
    
    obj["apiVersion"] = "platform.example.com/v1beta1"
    return obj


def convert_v1beta1_to_v1alpha1(obj: Dict[str, Any]) -> Dict[str, Any]:
    """Convert v1beta1 back to v1alpha1."""
    spec = obj.get("spec", {})
    
    if "resourceProfile" in spec:
        spec["size"] = spec.pop("resourceProfile")
    
    obj["apiVersion"] = "platform.example.com/v1alpha1"
    return obj


@app.post("/convert")
async def convert(request: Request):
    """Handle conversion webhook requests."""
    body = await request.json()
    
    desired_version = body["request"]["desiredAPIVersion"]
    objects = body["request"]["objects"]
    
    converted = []
    for obj in objects:
        current_version = obj["apiVersion"]
        
        if current_version == desired_version:
            converted.append(obj)
        elif current_version == "platform.example.com/v1alpha1" and \
             desired_version == "platform.example.com/v1beta1":
            converted.append(convert_v1alpha1_to_v1beta1(obj))
        elif current_version == "platform.example.com/v1beta1" and \
             desired_version == "platform.example.com/v1alpha1":
            converted.append(convert_v1beta1_to_v1alpha1(obj))
        else:
            return {
                "apiVersion": "apiextensions.k8s.io/v1",
                "kind": "ConversionReview",
                "response": {
                    "uid": body["request"]["uid"],
                    "result": {
                        "status": "Failed",
                        "message": f"Unknown conversion: {current_version} -> {desired_version}",
                    },
                },
            }
    
    return {
        "apiVersion": "apiextensions.k8s.io/v1",
        "kind": "ConversionReview",
        "response": {
            "uid": body["request"]["uid"],
            "result": {"status": "Success"},
            "convertedObjects": converted,
        },
    }

---

✅ Best Practices

API Design

1. Start declarative - State what you want, not how to get it

2. Use schemas - Strong typing catches errors early

3. Version from day one - Plan for evolution

4. Include status - Users need to know what's happening

5. Document with OpenAPI - Self-documenting APIs

Operators

1. Be idempotent - Same input = same output

2. Handle errors gracefully - Update status, don't crash

3. Use owner references - Enable garbage collection

4. Implement finalizers - Clean up external resources

5. Rate limit reconciliation - Avoid API throttling

Crossplane

1. Start with compositions - Build abstractions early

2. Use claims - Separate consumer from implementation

3. Test compositions - Validate before production

4. Monitor drift - Ensure desired state matches actual

---

🔗 What's Next?

In Article 10: Security and Compliance, we'll explore how to embed security into your platform using policy as code, shift-left security practices, and compliance automation.

---

📚 References

• Kubernetes Operator Pattern

• Crossplane Documentation

• Kopf - Kubernetes Operators Framework

• kubebuilder

• Operator SDK