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
Navigate to Edge Functions
Go to the Edge Functions section in your dashboard.
Create New Function
Click "New Function" and select Python as the language.
Write Your Code
Write your function logic in the code editor.
Deploy
Click "Deploy" to make your function available.
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
}Store API keys in environment variables, not in code.
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"}, 500Example: 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
| Package | Use Case |
|---|---|
httpx | HTTP client |
anthropic | Claude AI |
openai | OpenAI API |
pandas | Data processing |
numpy | Numerical computing |
pillow | Image processing |
pydantic | Data validation |
Request additional packages through the dashboard settings.