Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ Need to deploy your Worker to Cloudflare? Python Workers are in open beta and ha
- [**`websocket-stream-consumer/`**](websocket-stream-consumer) — shows how to use [WebSocket](https://developers.cloudflare.com/workers/runtime-apis/websockets/) to consume a stream of data with Python Workers.
- [**`chatroom/`**](chatroom) - A real-time chatroom using WebSocket.
- [**`sync-http-clients/`**](sync-http-clients) — demonstrates outbound HTTP with synchronous Python clients (`requests`, `urllib3`, and `httpx.Client`).
- [**`url-shortener/`**](url-shortener) — a URL shortener backed by Workers KV.
- [**`dynamic-py-py/`**](dynamic-py-py) — shows how to load and run a Python Worker dynamically at runtime using a [Worker Loader](https://developers.cloudflare.com/workers/runtime-apis/bindings/worker-loader/) binding.
- [**`django/`**](django) — runs a naive Django WSGI application directly on Python Workers.
- [**`django-todo-d1/`**](django-todo-d1) — implements the Todo-Backend API with Django and D1.
Expand Down
21 changes: 21 additions & 0 deletions tests/test_examples.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,27 @@ def test_sync_http_clients(dev_server):
assert result["saw_expected_text"] is True


def test_url_shortener(dev_server):
port = dev_server
base = f"http://localhost:{port}"
destination = "https://example.com/path"

response = requests.post(f"{base}/shorten", json={"url": destination})
assert response.status_code == 201
shortened = response.json()
assert len(shortened["code"]) == 8
assert shortened["short_url"] == f"{base}/{shortened['code']}"
assert shortened["url"] == destination

response = requests.get(shortened["short_url"], allow_redirects=False)
assert response.status_code == 302
assert response.headers["location"] == destination

response = requests.get(f"{base}/missing-code")
assert response.status_code == 404
assert response.json()["error"] == "not found"


def test_cron(dev_server):
port = dev_server
response = requests.get(f"http://localhost:{port}")
Expand Down
33 changes: 33 additions & 0 deletions url-shortener/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
# URL Shortener Example

[![Deploy to Cloudflare](https://deploy.workers.cloudflare.com/button)](https://deploy.workers.cloudflare.com/?url=https://github.com/cloudflare/python-workers-examples/tree/main/url-shortener)

A Python Worker that creates short URLs backed by Workers KV.

## Configure Workers KV

Create a namespace, then replace the placeholder ID in `wrangler.jsonc` with the ID from the command output:

```sh
uv run pywrangler kv namespace create LINKS
```

## Run locally

First ensure that [uv](https://docs.astral.sh/uv/getting-started/installation/#standalone-installer) is installed. Then run:

```sh
uv run pywrangler dev
curl -X POST http://localhost:8787/shorten \
-H 'content-type: application/json' \
-d '{"url":"https://example.com/path"}'
```

The response includes a `code`, `short_url`, and the original `url`. Request the returned short URL to receive a `302` redirect. Deploy with `uv run pywrangler deploy`.

## API

| Endpoint | Description |
|---|---|
| `POST /shorten` | Store an absolute HTTP(S) URL and return a short URL. |
| `GET /<code>` | Redirect to the stored URL with status `302`. |
13 changes: 13 additions & 0 deletions url-shortener/package.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
{
"name": "python-url-shortener",
"version": "0.0.0",
"private": true,
"scripts": {
"deploy": "uv run pywrangler deploy",
"dev": "uv run pywrangler dev",
"start": "uv run pywrangler dev"
},
"devDependencies": {
"wrangler": "^4.114.0"
}
}
13 changes: 13 additions & 0 deletions url-shortener/pyproject.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
[project]
name = "python-url-shortener"
version = "0.1.0"
description = "Workers KV URL shortener in Python"
readme = "README.md"
requires-python = ">=3.12"
dependencies = []

[dependency-groups]
dev = [
"workers-py",
"workers-runtime-sdk"
]
69 changes: 69 additions & 0 deletions url-shortener/src/entry.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
import uuid
from urllib.parse import urlsplit

from workers import Response, WorkerEntrypoint

MAX_CODE_ATTEMPTS = 5


def error(message, status, allow=None):
headers = {"Allow": allow} if allow else None
return Response.json({"error": message}, status=status, headers=headers)


def is_valid_destination(url):
try:
parsed = urlsplit(url)
hostname = parsed.hostname
_ = parsed.port
except ValueError:
return False
return parsed.scheme in ("http", "https") and bool(hostname)


class Default(WorkerEntrypoint):
async def fetch(self, request):
path = urlsplit(request.url).path

if path == "/shorten":
if request.method != "POST":
return error("method not allowed", 405, allow="POST")
return await self.shorten(request)

if request.method != "GET":
return error("method not allowed", 405, allow="GET")

code = path.lstrip("/")
if not code or "/" in code:
return error("not found", 404)
Comment on lines +36 to +38

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we only take the first component of the url here? So that https://thisworker.com/somecode/some/path works?


url = await self.env.LINKS.get(code)
if url is None:
return error("not found", 404)
return Response.redirect(url, 302)

async def shorten(self, request):
try:
body = await request.json()
url = body["url"]
except (KeyError, TypeError, ValueError):
return error("request body must contain a URL", 400)

if not isinstance(url, str) or not is_valid_destination(url):
return error("url must be an absolute HTTP(S) URL", 400)

for _ in range(MAX_CODE_ATTEMPTS):
code = uuid.uuid4().hex[:8]

if await self.env.LINKS.get(code) is not None:
# Code already exists, try again
continue

await self.env.LINKS.put(code, url)
origin = urlsplit(request.url)
short_url = f"{origin.scheme}://{origin.netloc}/{code}"
return Response.json(
{"code": code, "short_url": short_url, "url": url}, status=201
)

return error("could not allocate a short code", 503)
18 changes: 18 additions & 0 deletions url-shortener/wrangler.jsonc
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
{
"$schema": "node_modules/wrangler/config-schema.json",
"name": "python-url-shortener",
"main": "src/entry.py",
"compatibility_date": "2026-09-01",
"compatibility_flags": [
"python_workers"
],
"kv_namespaces": [
{
"binding": "LINKS",
"id": "<YOUR_KV_NAMESPACE_ID>"
}
],
"observability": {
"enabled": true
}
}
Loading