Skip to main content

​
Execution Environment

Cloud functions execute in a secure Python environment with access to these pre-injected globals — no imports needed:
GlobalDescription
reqRequest object (params, payload, headers, user)
dbDatabase service (queries, CRUD, relationships)
httpHTTP client (external API requests)
authAuthentication service (JWT tokens)
renderTemplate renderer (HTML with Jinja2)
use()Module system (import other functions)
queueBackground tasks (non-blocking execution)
configProject configuration and cron payloads
emailEmail service (send emails)

​
Request Object (req)

def main():
    # Query parameters (GET)
    name = req.get("name")
    age = req.get("age", 18)  # with default

    # Payload (POST/PUT)
    email = req.payload.get("email")
    data = req.payload  # full payload

    # Headers
    auth_header = req.headers.get("Authorization")

    # Authenticated user
    user_id = req.user["id"] if req.user else None
    user_email = req.user["email"] if req.user else None

    # HTTP method
    method = req.method  # GET, POST, PUT, DELETE

    # Project ID
    project_id = req.proj

    return {"name": name, "user": user_id}

​
Database (db)

# Query documents
result = db.query("users", limit=10)
users = result["data"]

# Query with filters
result = db.query("users", age_gte="18", limit=100)

# Find one document
user = db.find_one("users", id="user123")
user = db.find_one("users", email="john@example.com")

# Count documents
count = db.count_documents("users")
count = db.count_documents("users", status="active")

# Create document
doc = db.create_document("users", {
    "name": "John",
    "email": "john@example.com"
})

# Update document
updated = db.update_document("users", "user123", {
    "name": "John Doe"
})

# Delete document
deleted = db.delete_document("users", "user123")

​
HTTP Client (http)

# GET request
response = http.get("https://api.example.com/users")
users = response.json()

# GET with params and headers
response = http.get("https://api.example.com/users",
                   params={"limit": 10},
                   headers={"Authorization": "Bearer token"})

# POST request
response = http.post("https://api.example.com/users",
                    json={"name": "John"})

# PUT, PATCH, DELETE
response = http.put("https://api.example.com/users/123",
                   json={"name": "John Doe"})

response = http.patch("https://api.example.com/users/123",
                     json={"status": "active"})

response = http.delete("https://api.example.com/users/123")

# Response object
status = response.status_code
data = response.json()
text = response.text

​
Authentication (auth)

# Generate token for user
token = auth.create_token(
    user_id="user123",
    email="john@example.com",
    expires_in=3600  # 1 hour
)

# Verify token
token = req.headers.get("Authorization", "").replace("Bearer ", "")
try:
    payload = auth.verify_token(token)
    user_id = payload["user_id"]
except Exception as e:
    return {"error": "Invalid token"}, 401

​
Template Rendering (render)

# Render HTML string
def main():
    html = render.render_html(
        "<h1>Hello {{name}}!</h1>",
        context={"name": "John"}
    )
    return html

# Render template file
def main():
    html = render.render_template(
        "welcome.html",
        context={"name": "John", "email": "john@example.com"}
    )
    return html

​
Module System (use())

# functions/utils.py
def calculate_tax(amount):
    return amount * 0.08

def format_currency(amount):
    return f"${amount:.2f}"

# functions/checkout.py
def main():
    utils = use("utils")
    
    subtotal = 100.00
    tax = utils.calculate_tax(subtotal)
    total = subtotal + tax
    
    return {
        "subtotal": utils.format_currency(subtotal),
        "tax": utils.format_currency(tax),
        "total": utils.format_currency(total)
    }

​
Queue (queue)

def send_email(user_id):
    """Runs in background"""
    user = db.find_one("users", id=user_id)
    # send email...

def main():
    user_id = req.get("user_id")
    
    # Queue background task - returns immediately
    queue.add("send_email", user_id=user_id)
    
    # Call another deployed function
    queue.call_function("analytics-service",
                       event="signup",
                       user_id=user_id)
    
    return {"queued": True}

​
Configuration (config)

def main():
    api_key = config.get("api_key")
    environment = config.get("environment", "production")
    
    # Access cron job payload
    report_type = config.get("report_type")
    
    return {"environment": environment}

​
Available Standard Library Modules

  • json — JSON encoding/decoding
  • datetime, timedelta — Date and time
  • math — Mathematical functions
  • re — Regular expressions
  • random — Random number generation
  • secrets — Cryptographically secure random
  • time — Time utilities
  • hashlib — Hash functions
  • uuid — UUID generation
  • base64 — Base64 encoding/decoding
  • urllib — URL parsing
  • collections — Counter, defaultdict
  • itertools — Iterator functions
  • functools — Higher-order functions
  • decimal — Decimal precision
  • statistics — Statistical functions
  • dataclasses — Dataclasses

​
WebSocket Globals

WebSocket functions have additional objects: 
# session object
player_id = session.player_id
user = session.user
session.set("score", 100)
score = session.get("score")
latency = session.latency

# room object
room_id = room.id
room.broadcast({"event": "update", "data": 123})
room.send_to(player_id, {"message": "Hello"})
players = room.get_players()
count = room.get_player_count()
room.state["game_started"] = True
room.start_game_loop()
room.stop_game_loop()

# request object (WebSocket)
event_type = request.get("type")
position = request.get("position")
message = request.json()

​
Security & Sandboxing

Not available in cloud functions:
  • File system access (open, os.listdir)
  • System commands (os.system, subprocess)
  • Network sockets (use http instead)
  • Process manipulation
Available:
  • Safe built-ins (len, str, int, dict etc.)
  • Standard library modules listed above
  • HTTP requests via http object
  • Database access via db object

​
Timeouts

OperationTimeout
Function execution20 seconds
HTTP requests10 seconds
Database queriesNo explicit limit

​
Best Practices

# Always handle missing data with defaults
name = req.get("name", "Guest")

# Always check authentication when needed
if not req.user:
    return {"error": "Unauthorized"}, 401

# Always use try/except
try:
    result = risky_operation()
except Exception as e:
    return {"error": str(e)}, 500

# Return structured responses
return {
    "success": True,
    "data": {"id": 123},
    "timestamp": datetime.utcnow().isoformat()
}