from js import Response

async def on_fetch(request):
    return Response.new("Hello from Python on the edge!")

from js import Response, env

async def on_fetch(request):
    ai = env.AI

    response = await ai.run(
        "@cf/meta/llama-3-8b-instruct",
        messages=[{"role": "user", "content": "Explain quantum computing"}]
    )

    return Response.json(response)

# Generate embeddings and store them
embedding = await ai.run("@cf/baai/bge-base-en-v1.5", text="Python is awesome")
await env.VECTORIZE_INDEX.insert([{
    "id": "1",
    "values": embedding.data[0],
    "metadata": {"language": "python", "sentiment": "positive"}
}])

# That's it. Your documents are now queryable.
results = await env.AUTORAG.query("How do I implement async generators?")

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)

# Your boto3 code works as-is
import boto3

s3 = boto3.client('s3',
    endpoint_url='https://YOUR_ACCOUNT_ID.r2.cloudflarestorage.com',
    aws_access_key_id='YOUR_ACCESS_KEY',
    aws_secret_access_key='YOUR_SECRET_KEY'
)

# Upload that 10GB model file. Download it 1000 times. Pay $0 in bandwidth.
s3.upload_file('model.pkl', 'ml-artifacts', 'models/latest.pkl')

# Producer
async def on_fetch(request):
    await env.TASK_QUEUE.send({
        "task": "process_upload",
        "file_id": "abc123",
        "user_id": request.headers.get("X-User-ID")
    })
    return Response.new("Processing started", status=202)

# Consumer (another Worker)
async def queue_handler(batch):
    for message in batch.messages:
        await process_file(message.body["file_id"])
        message.ack()

async def on_fetch(request):
    browser = env.BROWSER

    # Screenshot any website, extract data, generate PDFs
    page = await browser.new_page()
    await page.goto("https://news.ycombinator.com")
    screenshot = await page.screenshot()

    return Response.new(screenshot, headers={"Content-Type": "image/png"})

# Before: postgresql://user:pass@db.region.rds.amazonaws.com/myapp
# After:  postgresql://user:pass@hyperdrive-id.hyperdrive.workers.dev/myapp

# That's it. Global connection pooling and caching. No code changes.

async def on_fetch(request):
    # Write millions of events
    await env.ANALYTICS.writeDataPoint({
        "index": ["user_id", "endpoint"],
        "dimension": [request.headers.get("X-User-ID"), request.url],
        "metric": {"response_time": 42, "status_code": 200}
    })

    # Query with SQL
    results = await env.ANALYTICS.query("""
        SELECT endpoint, AVG(response_time) as avg_latency
        FROM analytics
        WHERE timestamp > NOW() - INTERVAL '1 hour'
        GROUP BY endpoint
    """)

from js import Response, env
import json

async def on_fetch(request):
    if request.method == "POST":
        data = await request.json()

        # Generate embedding
        embedding = await env.AI.run(
            "@cf/baai/bge-base-en-v1.5",
            text=data["query"]
        )

        # Semantic search
        results = await env.VECTORIZE_INDEX.query(
            embedding.data[0],
            topK=5
        )

        # Generate response
        context = "\n".join([r.metadata.content for r in results.matches])
        response = await env.AI.run(
            "@cf/meta/llama-3-8b-instruct",
            messages=[
                {"role": "system", "content": f"Context: {context}"},
                {"role": "user", "content": data["query"]}
            ]
        )

        return Response.json({"answer": response.response})

# API endpoint triggers processing
async def on_fetch(request):
    file_url = (await request.json())["file_url"]

    # Store in R2
    file_data = await fetch(file_url)
    await env.BUCKET.put(f"uploads/{uuid4()}", file_data.body)

    # Queue for processing
    await env.PROCESS_QUEUE.send({
        "key": key,
        "user_id": request.headers.get("X-User-ID")
    })

    return Response.json({"status": "processing"})

# Queue consumer processes files
async def queue_handler(batch):
    for msg in batch.messages:
        # Get from R2
        obj = await env.BUCKET.get(msg.body["key"])
        data = await obj.text()

        # Process with AI
        summary = await env.AI.run(
            "@cf/facebook/bart-large-cnn",
            text=data
        )

        # Store results in D1
        await env.DB.prepare("""
            INSERT INTO summaries (user_id, summary, created_at)
            VALUES (?, ?, datetime('now'))
        """).bind(msg.body["user_id"], summary.summary).run()

        msg.ack()

# No separate task file needed!
async def on_fetch(request):
    await env.IMAGE_QUEUE.send({
        "url": "https://example.com/image.jpg"
    })
    return Response.new("Queued for processing")

async def queue_handler(batch):
    for msg in batch.messages:
        # Download, process, save to R2
        msg.ack()

import pandas as pd
import psycopg2
from sqlalchemy import create_engine

# Complex ETL pipeline
engine = create_engine('postgresql://...')
df = pd.read_sql("SELECT * FROM events WHERE...", engine)
df_grouped = df.groupby(['user_id', 'action']).agg({
    'duration': 'mean',
    'timestamp': 'count'
})
df_grouped.to_sql('analytics_summary', engine)

# Write events directly
await env.ANALYTICS.writeDataPoint({
    "dimension": [user_id, action],
    "metric": {"duration": duration}
})

# Query with SQL
results = await env.ANALYTICS.query("""
    SELECT
        user_id,
        action,
        AVG(duration) as avg_duration,
        COUNT(*) as count
    FROM analytics
    GROUP BY user_id, action
""")

# main.py - Your entire "stack" in one file
from js import Response, env
from fastapi import FastAPI
import asyncio

app = FastAPI()

# Serve static assets directly
@app.get("/static/{path:path}")
async def static(path: str):
    # Workers can serve from KV or R2
    asset = await env.ASSETS.get(path)
    return Response.new(asset.body, headers={
        "Content-Type": asset.httpMetadata.contentType
    })

# Your app logic
@app.post("/api/process")
async def process(data: dict):
    # Cache in KV (replaces Redis)
    await env.KV.put(f"cache:{data['id']}", data)

    # Queue background work (replaces Celery)
    await env.QUEUE.send(data)

    # Store in D1 (replaces Postgres)
    await env.DB.prepare(
        "INSERT INTO items (data) VALUES (?)"
    ).bind(data).run()

    # Track metrics (replaces Prometheus)
    await env.ANALYTICS.writeDataPoint({
        "dimension": ["process"],
        "metric": {"count": 1}
    })

    return {"status": "processed"}

# Scheduled tasks (replaces Celery Beat)
# Configure in wrangler.toml:
# [triggers]
# crons = ["0 * * * *"]

# Dockerfile
FROM python:3.11-slim as builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY . .
RUN apt-get update && apt-get install -y curl  # Health checks
EXPOSE 8000
CMD ["gunicorn", "app:app", "--bind", "0.0.0.0:8000", "--workers", "4"]

volumes:
  - ./static:/app/static  # Static files
  - ./uploads:/app/uploads  # User uploads
  - pgdata:/var/lib/postgresql/data  # Database
  - ./logs:/app/logs  # Log files

├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── app.py
└── nginx.conf

# Unlike containers, console.log goes to wrangler tail
async def on_fetch(request):
    console.log("Request received:", request.url)
    console.log("Headers:", dict(request.headers))

    try:
        result = await risky_operation()
        console.log("Success:", result)
    except Exception as e:
        console.error("Failed:", str(e))
        console.error("Stack trace:", traceback.format_exc())

    return Response.new("Done")

# debug_wrapper.py - Wrap your entire module
import sys
import traceback

try:
    # Your imports and code here
    from js import Response, env
    import fastapi

    async def on_fetch(request):
        return Response.new("Worker started successfully!")

except Exception as e:
    # This WILL show in deployment logs
    print(f"STARTUP ERROR: {e}", file=sys.stderr)
    print(f"TRACEBACK: {traceback.format_exc()}", file=sys.stderr)

    # Create a minimal handler that reports the error
    async def on_fetch(request):
        return Response.new(
            f"Worker failed to start: {str(e)}\n\n{traceback.format_exc()}",
            status=500,
            headers={"Content-Type": "text/plain"}
        )

# Start with the absolute minimum
async def on_fetch(request):
    return Response.new("Stage 1: Basic handler works")

# Then add imports one by one
from js import Response
async def on_fetch(request):
    return Response.new("Stage 2: Imports work")

# Then add bindings
from js import Response, env
async def on_fetch(request):
    try:
        # Test each binding
        kv_test = await env.KV.get("test")
        return Response.new(f"Stage 3: KV binding works: {kv_test}")
    except Exception as e:
        return Response.new(f"KV binding failed: {e}", status=500)

# Finally add your logic

# Use pdb in local development
import pdb

async def on_fetch(request):
    pdb.set_trace()  # Works in local dev!
    data = await request.json()
    return Response.json({"received": data})

from js import Response, env
import traceback
import json

async def on_fetch(request):
    try:
        # Your actual logic
        return await handle_request(request)

    except Exception as e:
        # In development, return detailed errors
        if env.ENVIRONMENT == "development":
            return Response.new(
                json.dumps({
                    "error": str(e),
                    "type": type(e).__name__,
                    "traceback": traceback.format_exc(),
                    "request": {
                        "url": str(request.url),
                        "method": request.method,
                        "headers": dict(request.headers)
                    }
                }, indent=2),
                status=500,
                headers={"Content-Type": "application/json"}
            )
        else:
            # In production, log details but return generic error
            console.error(f"Error: {e}", traceback.format_exc())
            return Response.new("Internal Server Error", status=500)

class Counter:
    def __init__(self, state, env):
        self.state = state
        self.env = env
        # Debug: Log initialization
        console.log(f"DO initialized with id: {state.id}")

    async def fetch(self, request):
        # Debug: Log every request
        console.log(f"DO fetch: {request.method} {request.url}")

        # Add debug endpoint
        if request.url.pathname == "/debug":
            # Return internal state for debugging
            storage_keys = await self.state.storage.list()
            return Response.json({
                "id": self.state.id.toString(),
                "storage_keys": list(storage_keys.keys()),
                "env_bindings": list(self.env.keys())
            })

        # Regular logic with error handling
        try:
            # Your logic here
            pass
        except Exception as e:
            console.error(f"DO error: {e}")
            return Response.new(f"DO Error: {e}", status=500)

# Capture and replay problematic requests
async def on_fetch(request):
    # Clone request for logging (body can only be read once)
    request_data = {
        "url": str(request.url),
        "method": request.method,
        "headers": dict(request.headers),
        "body": await request.text() if request.body else None
    }

    try:
        # Your logic
        response = await handle_request(request_data)
        return response
    except Exception as e:
        # Log the full request for replay
        console.error("Failed request:", json.dumps(request_data))
        console.error("Error:", str(e))

        # Store in KV for later debugging
        await env.DEBUG_REQUESTS.put(
            f"error_{datetime.now().isoformat()}",
            json.dumps({
                "request": request_data,
                "error": str(e),
                "traceback": traceback.format_exc()
            })
        )

        return Response.new("Error logged", status=500)

async def on_fetch(request):
    url = request.url

    # Add health/debug endpoints
    if url.pathname == "/health":
        # Test all bindings and dependencies
        checks = {}

        try:
            await env.KV.get("health_check")
            checks["kv"] = "ok"
        except:
            checks["kv"] = "failed"

        try:
            await env.DB.prepare("SELECT 1").first()
            checks["d1"] = "ok"
        except:
            checks["d1"] = "failed"

        try:
            await env.QUEUE.send({"test": True})
            checks["queue"] = "ok"
        except:
            checks["queue"] = "failed"

        status = 200 if all(v == "ok" for v in checks.values()) else 503

        return Response.json({
            "status": "healthy" if status == 200 else "unhealthy",
            "checks": checks,
            "timestamp": datetime.now().isoformat()
        }, status=status)

    # Regular routing
    return await handle_request(request)

import os
from cloudflare import Cloudflare

# Async support built-in
from cloudflare import AsyncCloudflare

client = Cloudflare(
    api_token=os.environ.get("CLOUDFLARE_API_TOKEN"),  # Recommended
    # or use api_email + api_key (legacy)
)

# Your IDE knows exactly what fields are available
zone = client.zones.create(
    account={"id": "your-account-id"},
    name="example.com",
    type="full",  # IDE autocompletes: "full" | "partial" | "secondary"
)

# Response objects are Pydantic models
print(zone.id)  # Fully typed!
print(zone.to_json(indent=2))  # Easy serialization
zone_dict = zone.to_dict()  # Convert to dict

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")

from cloudflare import Cloudflare

client = Cloudflare()

# Translate text using Meta's M2M model
result = client.ai.run(
    "@cf/meta/m2m100-1.2b",
    account_id="your-account-id",
    text="Hello, how are you?",
    source_lang="english",
    target_lang="spanish",
)

print(result)  # "Hola, ¿cómo estás?"

# Batch translation for efficiency
texts = ["Hello", "Goodbye", "Thank you"]
translations = []

for text in texts:
    result = client.ai.run(
        "@cf/meta/m2m100-1.2b",
        account_id="your-account-id",
        text=text,
        source_lang="english",
        target_lang="japanese",
    )
    translations.append(result)

from cloudflare import Cloudflare
from typing import List, Dict

client = Cloudflare()

def sync_dns_records(zone_id: str, desired_records: List[Dict]):
    """Declarative DNS management - define desired state"""

    # Get current records
    current_records = list(client.dns.records.list(zone_id=zone_id))

    # Create a map for easy lookup
    current_map = {(r.type, r.name): r for r in current_records}

    for desired in desired_records:
        key = (desired["type"], desired["name"])

        if key in current_map:
            # Update existing record
            record = current_map[key]
            client.dns.records.update(
                zone_id=zone_id,
                dns_record_id=record.id,
                **desired
            )
            print(f"Updated: {desired['name']} ({desired['type']})")
        else:
            # Create new record
            client.dns.records.create(
                zone_id=zone_id,
                **desired
            )
            print(f"Created: {desired['name']} ({desired['type']})")

# Define your DNS configuration
my_dns_config = [
    {
        "type": "A",
        "name": "api.example.com",
        "content": "192.0.2.1",
        "proxied": True,  # Enable Cloudflare proxy
    },
    {
        "type": "CNAME",
        "name": "www.example.com",
        "content": "example.com",
        "proxied": True,
    },
    {
        "type": "MX",
        "name": "example.com",
        "content": "mail.example.com",
        "priority": 10,
        "proxied": False,  # MX records can't be proxied
    },
]

sync_dns_records("your-zone-id", my_dns_config)

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",
)

class CloudflareKVStore:
    """Redis-like interface for Workers KV"""

    def __init__(self, client: Cloudflare, account_id: str, namespace_id: str):
        self.client = client
        self.account_id = account_id
        self.namespace_id = namespace_id

    def set(self, key: str, value: any, ttl: int = None):
        """Set a value with optional TTL"""
        metadata = {"timestamp": time.time()}

        self.client.kv.namespaces.values.update(
            account_id=self.account_id,
            namespace_id=self.namespace_id,
            key_name=key,
            value=json.dumps(value),
            metadata=json.dumps(metadata),
            expiration_ttl=ttl,
        )

    def get(self, key: str):
        """Get a value"""
        try:
            result = self.client.kv.namespaces.values.get(
                account_id=self.account_id,
                namespace_id=self.namespace_id,
                key_name=key,
            )
            return json.loads(result) if result else None
        except:
            return None

    def delete(self, key: str):
        """Delete a key"""
        self.client.kv.namespaces.values.delete(
            account_id=self.account_id,
            namespace_id=self.namespace_id,
            key_name=key,
        )

    def list_keys(self, prefix: str = None):
        """List all keys with optional prefix"""
        return list(self.client.kv.namespaces.keys.list(
            account_id=self.account_id,
            namespace_id=self.namespace_id,
            prefix=prefix,
        ))

# Use it like Redis/Memcached
kv = CloudflareKVStore(client, "your-account-id", namespace.id)

# Cache user session
kv.set("session:user123", {
    "user_id": "user123",
    "email": "user@example.com",
    "login_time": time.time(),
}, ttl=3600)  # 1 hour TTL

# Retrieve session
session = kv.get("session:user123")
print(session)

# List all sessions
for key in kv.list_keys(prefix="session:"):
    print(key.name)

from cloudflare import Cloudflare
import numpy as np
from typing import List, Dict

client = Cloudflare()

# Create a vector index
index = client.vectorize.indexes.create(
    account_id="your-account-id",
    name="product-search",
    dimensions=384,  # Using all-MiniLM-L6-v2 embeddings
    metric="cosine",
)

class SemanticSearch:
    """Semantic search engine using Vectorize"""

    def __init__(self, client: Cloudflare, account_id: str, index_name: str):
        self.client = client
        self.account_id = account_id
        self.index_name = index_name

    async def index_documents(self, documents: List[Dict[str, any]]):
        """Index documents with their embeddings"""
        vectors = []

        for doc in documents:
            # Generate embedding using Workers AI
            embedding_response = await self.client.ai.run(
                "@cf/baai/bge-base-en-v1.5",
                account_id=self.account_id,
                text=doc["content"],
            )

            vectors.append({
                "id": doc["id"],
                "values": embedding_response.data[0],
                "metadata": {
                    "title": doc["title"],
                    "content": doc["content"],
                    "category": doc.get("category", "general"),
                }
            })

        # Batch insert
        self.client.vectorize.indexes.insert(
            account_id=self.account_id,
            index_name=self.index_name,
            vectors=vectors,
        )

    async def search(self, query: str, top_k: int = 5):
        """Search for similar documents"""
        # Generate query embedding
        embedding_response = await self.client.ai.run(
            "@cf/baai/bge-base-en-v1.5",
            account_id=self.account_id,
            text=query,
        )

        # Search
        results = self.client.vectorize.indexes.query(
            account_id=self.account_id,
            index_name=self.index_name,
            vector=embedding_response.data[0],
            top_k=top_k,
            return_metadata=True,
        )

        return results.matches

# Usage
search_engine = SemanticSearch(client, "your-account-id", "product-search")

# Index your products
await search_engine.index_documents([
    {
        "id": "1",
        "title": "Python Cookbook",
        "content": "Advanced Python programming techniques and recipes",
        "category": "books",
    },
    {
        "id": "2",
        "title": "Machine Learning with Python",
        "content": "Comprehensive guide to ML algorithms using Python",
        "category": "books",
    },
])

# Search semantically
results = await search_engine.search("Python programming books")
for result in results:
    print(f"Score: {result.score:.3f} - {result.metadata['title']}")

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",
)

class D1Analytics:
    """Analytics database using D1"""

    def __init__(self, client: Cloudflare, account_id: str, database_id: str):
        self.client = client
        self.account_id = account_id
        self.database_id = database_id
        self._init_schema()

    def _init_schema(self):
        """Initialize database schema"""
        self.execute("""
            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
            )
        """)

        self.execute("""
            CREATE INDEX IF NOT EXISTS idx_user_events
            ON events(user_id, timestamp)
        """)

    def execute(self, query: str, params: List[Any] = None):
        """Execute a query"""
        return self.client.d1.database.query(
            account_id=self.account_id,
            database_id=self.database_id,
            sql=query,
            params=params or [],
        )

    def track_event(self, user_id: str, event_type: str, properties: Dict = None):
        """Track an analytics event"""
        import json

        self.execute(
            "INSERT INTO events (user_id, event_type, properties) VALUES (?, ?, ?)",
            [user_id, event_type, json.dumps(properties or {})]
        )

    def get_user_events(self, user_id: str, limit: int = 100):
        """Get events for a user"""
        result = self.execute(
            """
            SELECT * FROM events
            WHERE user_id = ?
            ORDER BY timestamp DESC
            LIMIT ?
            """,
            [user_id, limit]
        )
        return result.results

    def get_event_counts(self, start_date: str, end_date: str):
        """Get event counts by type"""
        result = self.execute(
            """
            SELECT
                event_type,
                COUNT(*) as count,
                COUNT(DISTINCT user_id) as unique_users
            FROM events
            WHERE timestamp BETWEEN ? AND ?
            GROUP BY event_type
            ORDER BY count DESC
            """,
            [start_date, end_date]
        )
        return result.results

    def to_dataframe(self, query_result) -> pd.DataFrame:
        """Convert query results to pandas DataFrame"""
        if not query_result.results:
            return pd.DataFrame()

        return pd.DataFrame(query_result.results)

# Usage
analytics = D1Analytics(client, "your-account-id", database.id)

# Track events
analytics.track_event("user123", "page_view", {"page": "/products"})
analytics.track_event("user123", "add_to_cart", {"product_id": "SKU-001"})

# Analyze with pandas
events_df = analytics.to_dataframe(
    analytics.get_event_counts("2024-01-01", "2024-12-31")
)
print(events_df.head())