Documentation Index
Fetch the complete documentation index at: https://docs.cocobase.buzz/llms.txt
Use this file to discover all available pages before exploring further.
Execution Environment
Cloud functions execute in a secure Python environment
with access to these pre-injected globals — no imports needed:
| Global | Description |
|---|
| req | Request object (params, payload, headers, user) |
| db | Database service (queries, CRUD, relationships) |
| http | HTTP client (external API requests) |
| auth | Authentication service (JWT tokens) |
| render | Template renderer (HTML with Jinja2) |
| use() | Module system (import other functions) |
| queue | Background tasks (non-blocking execution) |
| config | Project configuration and cron payloads |
| email | Email 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
| Operation | Timeout |
|---|
| Function execution | 20 seconds |
| HTTP requests | 10 seconds |
| Database queries | No 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()
}
