Script to use Cloudflare Workers with Python via Requests

workerskv.py

import os
import requests

def endpoint(key):
    accountID = os.getenv("WORKERS_KV_ACCOUNT_ID")
    namespaceID = os.getenv("WORKERS_KV_NAMESPACE_ID")
    return f"https://api.cloudflare.com/client/v4/accounts/{accountID}/storage/kv/namespaces/{namespaceID}/values/{key}"

def set(key, value, ttl=0, expiration=0):
    headers = {
        "X-Auth-Email": os.getenv("CLOUDFLARE_EMAIL"),
        "Authorization": f"Bearer {os.getenv('CLOUDFLARE_KV_API_TOKEN')}",
        "Content-Type": "application/json"
    }
    payload = [{"key": key, "value": value, "expiration_ttl": ttl, "expiration": expiration}]
    response = requests.put(
        f"https://api.cloudflare.com/client/v4/accounts/{os.getenv('WORKERS_KV_ACCOUNT_ID')}/storage/kv/namespaces/{os.getenv('WORKERS_KV_NAMESPACE_ID')}/bulk",
        headers=headers,
        json=payload
    )
    json_response = response.json()
    if not json_response["success"]:
        raise Exception(", ".join(json_response.get("errors", [])) or "Unknown error")
    return json_response["result"]

def get(key):
    headers = {
        "X-Auth-Email": os.getenv("CLOUDFLARE_EMAIL"),
        "Authorization": f"Bearer {os.getenv('CLOUDFLARE_KV_API_TOKEN')}",
        "Content-Type": "application/json"
    }
    response = requests.get(endpoint(key), headers=headers)
    json_response = response.json()
    if json_response.get("errors"):
        return None
    return response.text

def delete(key):
    headers = {
        "X-Auth-Email": os.getenv("CLOUDFLARE_EMAIL"),
        "Authorization": f"Bearer {os.getenv('CLOUDFLARE_KV_API_TOKEN')}",
        "Content-Type": "application/json"
    }
    response = requests.delete(endpoint(key), headers=headers)
    json_response = response.json()
    if not json_response["success"]:
        raise Exception(", ".join(json_response.get("errors", [])) or "Unknown error")
    return json_response["result"]