Use R2 from Workers

{
	"r2_buckets": [
		{
			"binding": "MY_BUCKET", // <~ valid JavaScript variable name
			"bucket_name": "<YOUR_BUCKET_NAME>"
		}
	]
}

[[r2_buckets]]
binding = "MY_BUCKET"
bucket_name = "<YOUR_BUCKET_NAME>"

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint<Env> {
  async fetch(request: Request) {
    const url = new URL(request.url);
    const key = url.pathname.slice(1);

    switch (request.method) {
      case "PUT": {
        await this.env.MY_BUCKET.put(key, request.body, {
          onlyIf: request.headers,
          httpMetadata: request.headers,
        });
        return new Response(`Put ${key} successfully!`);
      }
      case "GET": {
        const object = await this.env.MY_BUCKET.get(key, {
          onlyIf: request.headers,
          range: request.headers,
        });

        if (object === null) {
          return new Response("Object Not Found", { status: 404 });
        }

        const headers = new Headers();
        object.writeHttpMetadata(headers);
        headers.set("etag", object.httpEtag);

        // When no body is present, preconditions have failed
        return new Response("body" in object ? object.body : undefined, {
          status: "body" in object ? 200 : 412,
          headers,
        });
      }
      case "DELETE": {
        await this.env.MY_BUCKET.delete(key);
        return new Response("Deleted!");
      }
      default:
        return new Response("Method Not Allowed", {
          status: 405,
          headers: {
            Allow: "PUT, GET, DELETE",
          },
        });
    }
  }
};

export default {
  async fetch(request, env) {
    const url = new URL(request.url);
    const key = url.pathname.slice(1);

    switch (request.method) {
      case "PUT": {
        await env.MY_BUCKET.put(key, request.body, {
          onlyIf: request.headers,
          httpMetadata: request.headers,
        });
        return new Response(`Put ${key} successfully!`);
      }
      case "GET": {
        const object = await env.MY_BUCKET.get(key, {
          onlyIf: request.headers,
          range: request.headers,
        });

        if (object === null) {
          return new Response("Object Not Found", { status: 404 });
        }

        const headers = new Headers();
        object.writeHttpMetadata(headers);
        headers.set("etag", object.httpEtag);

        // When no body is present, preconditions have failed
        return new Response("body" in object ? object.body : undefined, {
          status: "body" in object ? 200 : 412,
          headers,
        });
      }
      case "DELETE": {
        await env.MY_BUCKET.delete(key);
        return new Response("Deleted!");
      }
      default:
        return new Response("Method Not Allowed", {
          status: 405,
          headers: {
            Allow: "PUT, GET, DELETE",
          },
        });
    }
  }
}

from workers import WorkerEntrypoint, Response
from urllib.parse import urlparse


class Default(WorkerEntrypoint):
	async def fetch(self, request):
		url = urlparse(request.url)
		key = url.path[1:]

		if request.method == "PUT":
			await self.env.MY_BUCKET.put(
				key,
				request.body,
				onlyIf=request.headers,
				httpMetadata=request.headers,
			)
			return Response(f"Put {key} successfully!")
		elif request.method == "GET":
			obj = await self.env.MY_BUCKET.get(
				key,
				onlyIf=request.headers,
				range=request.headers,
			)

			if obj is None:
				return Response("Object Not Found", status=404)

			# When no body is present, preconditions have failed
			body = obj.body if hasattr(obj, "body") else None
			status = 200 if hasattr(obj, "body") else 412

			headers = {"etag": obj.httpEtag}
			return Response(body, status=status, headers=headers)
		elif request.method == "DELETE":
			await self.env.MY_BUCKET.delete(key)
			return Response("Deleted!")
		else:
			return Response(
				"Method Not Allowed",
				status=405,
				headers={"Allow": "PUT, GET, DELETE"},
			)

const ALLOW_LIST = ["cat-pic.jpg"];

// Check requests for a pre-shared secret
const hasValidHeader = (request, env) => {
	return request.headers.get("X-Custom-Auth-Key") === env.AUTH_KEY_SECRET;
};

function authorizeRequest(request, env, key) {
	switch (request.method) {
		case "PUT":
		case "DELETE":
			return hasValidHeader(request, env);
		case "GET":
			return ALLOW_LIST.includes(key);
		default:
			return false;
	}
}

export default {
	async fetch(request, env, ctx) {
		const url = new URL(request.url);
		const key = url.pathname.slice(1);

		if (!authorizeRequest(request, env, key)) {
			return new Response("Forbidden", { status: 403 });
		}

		// ...
	},
};

from workers import WorkerEntrypoint, Response
from urllib.parse import urlparse

ALLOW_LIST = ["cat-pic.jpg"]

# Check requests for a pre-shared secret
def has_valid_header(request, env):
	return request.headers.get("X-Custom-Auth-Key") == env.AUTH_KEY_SECRET

def authorize_request(request, env, key):
	if request.method in ["PUT", "DELETE"]:
		return has_valid_header(request, env)
	elif request.method == "GET":
		return key in ALLOW_LIST
	else:
		return False

class Default(WorkerEntrypoint):
	async def fetch(self, request):
		url = urlparse(request.url)
		key = url.path[1:]

		if not authorize_request(request, self.env, key):
			return Response("Forbidden", status=403)

		# ...