Patterns for using KV in your edge functions.Documentation Index
Fetch the complete documentation index at: https://developers.telnyx.com/llms.txt
Use this file to discover all available pages before exploring further.
Session Storage with TTL
Store user sessions at the edge with automatic expiration:- JavaScript
- Python
- Go
const SESSION_TTL = 86400; // 24 hours
async function handler(request) {
const sessionId = request.headers.get("X-Session-ID");
// Get session
let session = await kvGet(`session:${sessionId}`);
if (!session) {
session = JSON.stringify({ created: Date.now(), views: 0 });
} else {
const data = JSON.parse(session);
data.views++;
session = JSON.stringify(data);
}
// Update session with TTL (auto-expires in 24h)
await kvPutWithTTL(`session:${sessionId}`, session, SESSION_TTL);
return new Response(session);
}
// Helper: PUT with TTL
async function kvPutWithTTL(key, value, ttl) {
const response = await fetch(
`https://api.telnyx.com/v2/storage/kvs/${KV_NAMESPACE_ID}/keys/${encodeURIComponent(key)}`,
{
method: "PUT",
headers: {
"Authorization": `Bearer ${API_KEY}`,
"Content-Type": "application/json"
},
body: JSON.stringify({
value: toBase64(value),
expiration_ttl: ttl
})
}
);
if (!response.ok) {
throw new Error(`KV write error: ${response.status}`);
}
}
import os, base64, json, time, httpx
from urllib.parse import quote
SESSION_TTL = 86400 # 24 hours
async def kv_put_with_ttl(key: str, value: str, ttl: int):
async with httpx.AsyncClient() as client:
await client.put(
f"https://api.telnyx.com/v2/storage/kvs/{os.getenv('KV_NAMESPACE_ID')}/keys/{quote(key, safe='')}",
headers={"Authorization": f"Bearer {os.getenv('TELNYX_API_KEY')}", "Content-Type": "application/json"},
json={"value": base64.b64encode(value.encode()).decode(), "expiration_ttl": ttl}
)
async def handler(request):
session_id = request.headers.get("X-Session-ID")
session = await kv_get(f"session:{session_id}")
if not session:
session = json.dumps({"created": int(time.time()), "views": 0})
else:
data = json.loads(session)
data["views"] += 1
session = json.dumps(data)
await kv_put_with_ttl(f"session:{session_id}", session, SESSION_TTL)
return {"body": session}
const sessionTTL = 86400 // 24 hours
func handler(w http.ResponseWriter, r *http.Request) {
sessionID := r.Header.Get("X-Session-ID")
session, _ := kvGet(fmt.Sprintf("session:%s", sessionID))
var data map[string]interface{}
if session == "" {
data = map[string]interface{}{"created": time.Now().Unix(), "views": 0}
} else {
json.Unmarshal([]byte(session), &data)
data["views"] = data["views"].(float64) + 1
}
updated, _ := json.Marshal(data)
kvPutWithTTL(fmt.Sprintf("session:%s", sessionID), string(updated), sessionTTL)
w.Header().Set("Content-Type", "application/json")
w.Write(updated)
}
API Response Caching with TTL
Cache expensive API responses with automatic expiration:- JavaScript
- Python
- Go
const CACHE_TTL = 300; // 5 minutes
async function handler(request) {
const cacheKey = `api:${new URL(request.url).pathname}`;
// Check cache
const cached = await kvGet(cacheKey);
if (cached) {
return new Response(cached, {
headers: { "X-Cache": "HIT" }
});
}
// Fetch from origin
const response = await fetch("https://api.example.com/data");
const data = await response.text();
// Cache with 5-minute TTL
await kvPutWithTTL(cacheKey, data, CACHE_TTL);
return new Response(data, {
headers: { "X-Cache": "MISS" }
});
}
CACHE_TTL = 300 # 5 minutes
async def handler(request):
from urllib.parse import urlparse
cache_key = f"api:{urlparse(request.url).path}"
cached = await kv_get(cache_key)
if cached:
return {"body": cached, "headers": {"X-Cache": "HIT"}}
async with httpx.AsyncClient() as client:
resp = await client.get("https://api.example.com/data")
data = resp.text
await kv_put_with_ttl(cache_key, data, CACHE_TTL)
return {"body": data, "headers": {"X-Cache": "MISS"}}
const cacheTTL = 300 // 5 minutes
func handler(w http.ResponseWriter, r *http.Request) {
cacheKey := fmt.Sprintf("api:%s", r.URL.Path)
cached, _ := kvGet(cacheKey)
if cached != "" {
w.Header().Set("X-Cache", "HIT")
w.Write([]byte(cached))
return
}
resp, _ := http.Get("https://api.example.com/data")
defer resp.Body.Close()
data, _ := io.ReadAll(resp.Body)
kvPutWithTTL(cacheKey, string(data), cacheTTL)
w.Header().Set("X-Cache", "MISS")
w.Write(data)
}
Feature Flags
Store and retrieve feature flags:- JavaScript
- Python
- Go
async function handler(request) {
const newUIEnabled = await kvGet("feature:new-ui");
if (newUIEnabled === "true") {
return serveNewUI(request);
}
return serveOldUI(request);
}
async def handler(request):
enabled = await kv_get("feature:new-ui")
if enabled == "true":
return serve_new_ui(request)
return serve_old_ui(request)
func handler(w http.ResponseWriter, r *http.Request) {
enabled, _ := kvGet("feature:new-ui")
if enabled == "true" {
serveNewUI(w, r)
return
}
serveOldUI(w, r)
}