Python Functions

Build powerful serverless functions with Python for AI, data processing, and business logic.

Overview

Python Edge Functions are ideal for:

• AI and machine learning tasks
• Data processing and transformation
• Complex business logic
• Third-party API integrations
• Webhooks and event handling

Creating a Function

Function Structure

# functions/hello.py

async def handler(request):
    """
    Main handler function that receives the request.

    Args:
        request: The incoming request object

    Returns:
        dict or tuple: Response data or (data, status_code)
    """
    # Get request data
    body = await request.json()
    name = body.get('name', 'World')

    # Return response
    return {
        "message": f"Hello, {name}!",
        "status": "success"
    }

Request Object

async def handler(request):
    # HTTP method
    method = request.method  # 'GET', 'POST', etc.

    # Headers
    auth = request.headers.get('authorization')
    content_type = request.headers.get('content-type')

    # Query parameters
    page = request.query.get('page', '1')
    limit = request.query.get('limit', '10')

    # URL parameters (from route)
    user_id = request.params.get('id')

    # Request body (JSON)
    if method == 'POST':
        body = await request.json()

    # Form data
    form = await request.form()

    # Raw body
    raw = await request.body()

    # Authenticated user (if available)
    user = request.user

    return {"method": method}

Response Formats

# Simple dict response (200 OK)
async def handler(request):
    return {"message": "Success"}

# Response with status code
async def handler(request):
    return {"error": "Not found"}, 404

# Response with headers
async def handler(request):
    return {
        "body": {"data": "value"},
        "status": 201,
        "headers": {
            "X-Custom-Header": "value"
        }
    }

Database Access

from dypai import db

async def handler(request):
    # Query data
    users = await db.query("SELECT * FROM users WHERE active = $1", [True])

    # Insert data
    result = await db.query(
        "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id",
        ["user@example.com", "John Doe"]
    )
    new_id = result[0]['id']

    # Update data
    await db.query(
        "UPDATE users SET last_login = NOW() WHERE id = $1",
        [user_id]
    )

    # Delete data
    await db.query("DELETE FROM sessions WHERE expired_at < NOW()")

    return {"users": users}

Storage Access

from dypai import storage

async def handler(request):
    # Upload file
    file = await request.form()
    uploaded = await storage.upload(
        bucket='documents',
        path='reports/report.pdf',
        file=file['file'],
        content_type='application/pdf'
    )

    # Get signed URL
    url = await storage.create_signed_url(
        bucket='documents',
        path='reports/report.pdf',
        expires_in=3600
    )

    # Download file
    content = await storage.download(
        bucket='documents',
        path='reports/report.pdf'
    )

    # Delete file
    await storage.delete(
        bucket='documents',
        path='reports/old-report.pdf'
    )

    return {"url": url}

Making HTTP Requests

import httpx

async def handler(request):
    async with httpx.AsyncClient() as client:
        # GET request
        response = await client.get(
            'https://api.example.com/data',
            headers={'Authorization': 'Bearer token'}
        )
        data = response.json()

        # POST request
        response = await client.post(
            'https://api.example.com/webhook',
            json={'event': 'user_created', 'user_id': 123}
        )

    return {"external_data": data}

AI Integration

from anthropic import Anthropic

async def handler(request):
    body = await request.json()
    prompt = body.get('prompt')

    client = Anthropic()
    message = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )

    return {
        "response": message.content[0].text
    }

Environment Variables

import os

async def handler(request):
    # Access environment variables
    api_key = os.environ.get('EXTERNAL_API_KEY')
    debug_mode = os.environ.get('DEBUG', 'false') == 'true'

    return {"debug": debug_mode}

Error Handling

async def handler(request):
    try:
        body = await request.json()
        user_id = body.get('user_id')

        if not user_id:
            return {"error": "user_id is required"}, 400

        # Process request
        result = await process_user(user_id)
        return {"result": result}

    except ValueError as e:
        return {"error": str(e)}, 400

    except Exception as e:
        # Log error for debugging
        print(f"Error: {e}")
        return {"error": "Internal server error"}, 500

Example: Webhook Processor

import hmac
import hashlib
from dypai import db

async def handler(request):
    # Verify webhook signature
    signature = request.headers.get('x-webhook-signature')
    body = await request.body()

    expected = hmac.new(
        os.environ['WEBHOOK_SECRET'].encode(),
        body,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return {"error": "Invalid signature"}, 401

    # Process webhook
    data = await request.json()
    event_type = data.get('type')

    if event_type == 'payment.completed':
        await db.query(
            "UPDATE orders SET status = 'paid' WHERE payment_id = $1",
            [data['payment_id']]
        )

    elif event_type == 'user.created':
        await send_welcome_email(data['user']['email'])

    return {"received": True}

Available Packages

Next steps

• Learn Rust functions
• Connect to endpoints
• AI integration