Serverless APIs

# app/main.py - FastAPI application with Lambda handler

from fastapi import FastAPI, HTTPException
from mangum import Mangum
from pydantic import BaseModel

app = FastAPI(title="Serverless API")

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float

items_db = {}

@app.get("/")
async def root():
    return {"message": "Hello from Lambda!", "platform": "AWS"}

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

@app.post("/items/")
async def create_item(item: Item):
    item_id = len(items_db) + 1
    items_db[item_id] = item
    return {"id": item_id, **item.model_dump()}

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

# Lambda handler - Mangum wraps FastAPI app
# lifespan="off" disables startup/shutdown events (not needed in Lambda)
handler = Mangum(app, lifespan="off")

# When deployed to Lambda:
# - API Gateway invokes handler(event, context)
# - Mangum translates event → FastAPI request
# - FastAPI processes and returns response
# - Mangum translates response → API Gateway format

# template.yaml - SAM template

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Serverless FastAPI application

Globals:
  Function:
    Timeout: 30
    MemorySize: 512
    Runtime: python3.11
    Environment:
      Variables:
        ENVIRONMENT: production
        LOG_LEVEL: INFO

Resources:
  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./
      Handler: app.main.handler
      Events:
        # Catch-all route for all paths
        ApiEvent:
          Type: HttpApi
          Properties:
            Path: /{proxy+}
            Method: ANY
        # Root path
        RootEvent:
          Type: HttpApi
          Properties:
            Path: /
            Method: ANY
      Policies:
        # Grant DynamoDB access
        - DynamoDBCrudPolicy:
            TableName: !Ref ItemsTable
        # Grant Systems Manager Parameter Store access
        - SSMParameterReadPolicy:
            ParameterName: api/*

  # DynamoDB table for data persistence
  ItemsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: items
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH

Outputs:
  ApiUrl:
    Description: API Gateway endpoint URL
    Value: !Sub "https://${ServerlessHttpApi}.execute-api.${AWS::Region}.amazonaws.com/"

  FunctionArn:
    Description: Lambda Function ARN
    Value: !GetAtt ApiFunction.Arn

# main.py - Google Cloud Function with HTTP trigger

import functions_framework
from flask import jsonify, Request
import json

# In-memory storage (use Firestore in production)
items_db = {}

@functions_framework.http
def api_handler(request: Request):
    """HTTP Cloud Function entry point"""

    # CORS headers (allow requests from browser)
    headers = {
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
        'Access-Control-Allow-Headers': 'Content-Type'
    }

    # Handle OPTIONS for CORS preflight
    if request.method == 'OPTIONS':
        return ('', 204, headers)

    # Route requests
    path = request.path
    method = request.method

    # GET /
    if path == '/' and method == 'GET':
        return (jsonify({
            "message": "Hello from Google Cloud Functions!",
            "platform": "Google Cloud"
        }), 200, headers)

    # GET /items
    elif path == '/items' and method == 'GET':
        return (jsonify(list(items_db.values())), 200, headers)

    # POST /items
    elif path == '/items' and method == 'POST':
        data = request.get_json()
        item_id = len(items_db) + 1
        item = {"id": item_id, **data}
        items_db[item_id] = item
        return (jsonify(item), 201, headers)

    # GET /items/{id}
    elif path.startswith('/items/') and method == 'GET':
        item_id = int(path.split('/')[-1])
        if item_id in items_db:
            return (jsonify(items_db[item_id]), 200, headers)
        return (jsonify({"error": "Not found"}), 404, headers)

    # 404 for unknown routes
    return (jsonify({"error": "Not found"}), 404, headers)

# main.py - FastAPI on Google Cloud Functions

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import functions_framework

app = FastAPI(title="Cloud Functions API")

class Item(BaseModel):
    name: str
    price: float

items_db = {}

@app.get("/")
async def root():
    return {"message": "FastAPI on Google Cloud!", "platform": "Google Cloud"}

@app.post("/items/")
async def create_item(item: Item):
    item_id = len(items_db) + 1
    items_db[item_id] = item.model_dump()
    return {"id": item_id, **item.model_dump()}

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Not found")
    return items_db[item_id]

# Wrapper to convert Cloud Functions request → ASGI
@functions_framework.http
def handler(request):
    from starlette.requests import Request as StarletteRequest
    from starlette.responses import Response
    import asyncio

    # Convert Flask request to ASGI scope
    scope = {
        "type": "http",
        "method": request.method,
        "path": request.path,
        "query_string": request.query_string,
        "headers": [(k.lower().encode(), v.encode()) for k, v in request.headers.items()],
    }

    # Run FastAPI asynchronously
    async def receive():
        return {"type": "http.request", "body": request.get_data()}

    response_started = False
    status_code = 200
    response_headers = []
    body_parts = []

    async def send(message):
        nonlocal response_started, status_code, response_headers, body_parts
        if message["type"] == "http.response.start":
            response_started = True
            status_code = message["status"]
            response_headers = message.get("headers", [])
        elif message["type"] == "http.response.body":
            body_parts.append(message.get("body", b""))

    # Execute FastAPI
    asyncio.run(app(scope, receive, send))

    # Return Flask response
    headers = {k.decode(): v.decode() for k, v in response_headers}
    return (b"".join(body_parts), status_code, headers)

# requirements.txt
# fastapi
# functions-framework

# Deploy command
gcloud functions deploy my-api \
  --runtime python311 \
  --trigger-http \
  --allow-unauthenticated \
  --entry-point api_handler \
  --region us-central1 \
  --memory 512MB \
  --timeout 60s

# View logs
gcloud functions logs read my-api --limit 50

# Test locally first
functions-framework --target=api_handler --debug

# function_app.py - Azure Functions with HTTP trigger (v2 programming model)

import azure.functions as func
import json
import logging

app = func.FunctionApp()

items_db = {}

@app.function_name(name="health")
@app.route(route="health", methods=["GET"])
def health(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Health check endpoint called')
    return func.HttpResponse(
        json.dumps({"status": "healthy", "platform": "Azure"}),
        mimetype="application/json",
        status_code=200
    )

@app.function_name(name="create_item")
@app.route(route="items", methods=["POST"])
def create_item(req: func.HttpRequest) -> func.HttpResponse:
    try:
        req_body = req.get_json()
        item_id = len(items_db) + 1
        item = {"id": item_id, **req_body}
        items_db[item_id] = item

        return func.HttpResponse(
            json.dumps(item),
            mimetype="application/json",
            status_code=201
        )
    except ValueError:
        return func.HttpResponse(
            json.dumps({"error": "Invalid JSON"}),
            mimetype="application/json",
            status_code=400
        )

@app.function_name(name="get_item")
@app.route(route="items/{item_id:int}", methods=["GET"])
def get_item(req: func.HttpRequest) -> func.HttpResponse:
    item_id = int(req.route_params.get('item_id'))

    if item_id in items_db:
        return func.HttpResponse(
            json.dumps(items_db[item_id]),
            mimetype="application/json",
            status_code=200
        )

    return func.HttpResponse(
        json.dumps({"error": "Item not found"}),
        mimetype="application/json",
        status_code=404
    )

@app.function_name(name="list_items")
@app.route(route="items", methods=["GET"])
def list_items(req: func.HttpRequest) -> func.HttpResponse:
    return func.HttpResponse(
        json.dumps(list(items_db.values())),
        mimetype="application/json",
        status_code=200
    )

# Create function app (Consumption plan - serverless)
az functionapp create \
  --resource-group myResourceGroup \
  --consumption-plan-location eastus \
  --runtime python \
  --runtime-version 3.11 \
  --functions-version 4 \
  --name my-api-function \
  --storage-account mystorageaccount \
  --os-type Linux

# Deploy code
func azure functionapp publish my-api-function

# View logs
func azure functionapp logstream my-api-function

# Using VS Code Azure Functions extension (recommended):
# - Install "Azure Functions" extension
# - Right-click function_app.py → "Deploy to Function App"
# - Select or create function app
# - Deployment handles everything automatically

# Install Serverless Framework globally
npm install -g serverless

# Create new Python service
serverless create --template aws-python3 --path my-api
cd my-api

# Or use interactive setup
serverless

# Install Python plugin for better packaging
serverless plugin install -n serverless-python-requirements

# serverless.yml - FastAPI application on AWS Lambda

service: my-api

frameworkVersion: '3'

provider:
  name: aws
  runtime: python3.11
  region: us-east-1
  stage: ${opt:stage, 'dev'}
  memorySize: 512
  timeout: 30
  environment:
    STAGE: ${self:provider.stage}
    TABLE_NAME: ${self:service}-${self:provider.stage}-items
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - dynamodb:Query
            - dynamodb:Scan
            - dynamodb:GetItem
            - dynamodb:PutItem
            - dynamodb:UpdateItem
            - dynamodb:DeleteItem
          Resource:
            - !GetAtt ItemsTable.Arn

functions:
  api:
    handler: app.main.handler
    events:
      - httpApi:
          path: /{proxy+}
          method: ANY
      - httpApi:
          path: /
          method: ANY

resources:
  Resources:
    ItemsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: ${self:provider.environment.TABLE_NAME}
        BillingMode: PAY_PER_REQUEST
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH

plugins:
  - serverless-python-requirements

custom:
  pythonRequirements:
    dockerizePip: true
    slim: true
    strip: false

# app/main.py

from fastapi import FastAPI, HTTPException
from mangum import Mangum
from pydantic import BaseModel
import os

app = FastAPI(
    title="Serverless API",
    version="1.0.0",
    root_path=f"/{os.getenv('STAGE', 'dev')}"  # For API Gateway stages
)

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float

@app.get("/")
async def root():
    return {
        "message": "Serverless Framework API",
        "stage": os.getenv("STAGE"),
        "platform": "AWS Lambda"
    }

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

@app.post("/items/")
async def create_item(item: Item):
    # In production, save to DynamoDB using boto3
    return {"id": 1, **item.model_dump()}

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    # In production, fetch from DynamoDB
    return {"id": item_id, "name": "Example"}

# Lambda handler
handler = Mangum(app, lifespan="off")

# Deploy to dev stage (default)
serverless deploy

# Deploy to production
serverless deploy --stage prod

# Deploy only function (faster for code updates)
serverless deploy function -f api

# View logs
serverless logs -f api --tail

# Invoke function locally
serverless invoke local -f api --data '{"httpMethod":"GET","path":"/"}'

# Remove all resources
serverless remove

# Info about deployed service
serverless info

# Environment variables
serverless deploy --stage prod --param="dbHost=prod-db.example.com"