Storage: D1, R2, KV & Hyperdrive

Cloudflare offers a complete storage stack for Python developers — from edge SQL databases to globally distributed key-value stores. Replace your entire data layer with zero egress fees.

D1: SQLite at the Edge

D1 is Cloudflare's serverless SQL database built on SQLite. It's globally distributed with automatic backups and 30-day point-in-time recovery (Time Travel).

from js import Response, env

async def on_fetch(request):
    db = env.DB  # It's just SQLite, but globally distributed

    results = await db.prepare("""
        SELECT user_id, COUNT(*) as request_count
        FROM api_logs
        WHERE timestamp > datetime('now', '-1 hour')
        GROUP BY user_id
    """).all()

    return Response.json(results)

Schema & Queries

# Create tables
await env.DB.prepare("""
    CREATE TABLE IF NOT EXISTS events (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        user_id TEXT NOT NULL,
        event_type TEXT NOT NULL,
        properties TEXT,
        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
    )
""").run()

# Create indexes
await env.DB.prepare("""
    CREATE INDEX IF NOT EXISTS idx_user_events
    ON events(user_id, timestamp)
""").run()

# Insert with prepared statements
await env.DB.prepare(
    "INSERT INTO events (user_id, event_type, properties) VALUES (?, ?, ?)"
).bind(user_id, event_type, json.dumps(properties)).run()

# Query with parameters
result = await env.DB.prepare("""
    SELECT * FROM events
    WHERE user_id = ?
    ORDER BY timestamp DESC
    LIMIT ?
""").bind(user_id, 100).all()

D1 via the Python SDK

from cloudflare import Cloudflare
import pandas as pd
from typing import List, Dict, Any

client = Cloudflare()

# Create a D1 database
database = client.d1.database.create(
    account_id="your-account-id",
    name="analytics",
)

# Execute queries via the SDK
client.d1.database.query(
    account_id="your-account-id",
    database_id="your-database-id",
    sql="INSERT INTO events (user_id, event_type) VALUES (?, ?)",
    params=["user123", "page_view"],
)

R2: S3-Compatible Object Storage

R2 provides S3-compatible object storage with zero egress fees. Use your existing boto3 code — just change the endpoint.

R2 in Python Workers

from js import Response, env

async def on_fetch(request):
    # Store a file
    await env.BUCKET.put("uploads/file.txt", request.body)

    # Retrieve a file
    object = await env.BUCKET.get("uploads/file.txt")
    data = await object.text()

    # Serve with proper content type
    return Response.new(object.body, headers={
        "Content-Type": object.httpMetadata.contentType
    })

R2 via the Python SDK (boto3 Compatible)

from cloudflare import Cloudflare
import boto3
from datetime import timedelta

client = Cloudflare()

# Create an R2 bucket
bucket = client.r2.buckets.create(
    account_id="your-account-id",
    name="ml-models",
    storage_class="Standard",
)

# Get temporary S3-compatible credentials
creds = client.r2.temporary_credentials.create(
    account_id="your-account-id",
    bucket=bucket.name,
    permission="read_write",
    ttl=timedelta(hours=1),
)

# Use with boto3 — zero code changes!
s3 = boto3.client(
    "s3",
    endpoint_url=f"https://{creds.account_id}.r2.cloudflarestorage.com",
    aws_access_key_id=creds.access_key_id,
    aws_secret_access_key=creds.secret_access_key,
    aws_session_token=creds.session_token,
)

# Upload your ML model
s3.upload_file("model.pkl", bucket.name, "models/latest.pkl")

# List objects — no egress fees!
for obj in s3.list_objects_v2(Bucket=bucket.name)["Contents"]:
    print(f"{obj['Key']}: {obj['Size']} bytes")

Workers KV: Global Key-Value Store

Workers KV is a globally distributed key-value store — think Redis, but with no server management and global replication built in.

KV in Python Workers

from js import Response, env

async def on_fetch(request):
    # Set a value
    await env.KV.put("session:user123", json.dumps({"role": "admin"}))

    # Get a value
    value = await env.KV.get("session:user123")

    # Set with expiration (TTL in seconds)
    await env.KV.put("cache:results", data, expirationTtl=3600)

    return Response.json({"session": value})

KV via the Python SDK

from cloudflare import Cloudflare
import json
import time

client = Cloudflare()

# Create a KV namespace
namespace = client.kv.namespaces.create(
    account_id="your-account-id",
    title="user-sessions",
)

# Set a value with optional TTL
client.kv.namespaces.values.update(
    account_id="your-account-id",
    namespace_id=namespace.id,
    key_name="user:123",
    value=json.dumps({"role": "admin"}),
    metadata=json.dumps({"timestamp": time.time()}),
    expiration_ttl=3600,
)

# Get a value
result = client.kv.namespaces.values.get(
    account_id="your-account-id",
    namespace_id=namespace.id,
    key_name="user:123",
)

Hyperdrive: Accelerate Existing Databases

Already have a PostgreSQL or MySQL database? Hyperdrive accelerates connections to your existing database by up to 10x — keep your ORM, get edge performance.

# Keep your SQLAlchemy code — Hyperdrive accelerates the connection
# wrangler.toml
# [[hyperdrive]]
# binding = "HYPERDRIVE"
# id = "your-hyperdrive-id"

from js import env

async def on_fetch(request):
    # Hyperdrive provides an accelerated connection string
    connection_string = env.HYPERDRIVE.connectionString

    # Use with SQLAlchemy, psycopg2, or any DB driver
    # Connection pooling and caching handled automatically

Migration Reference

Map your existing storage stack to Cloudflare equivalents: