Skip to content

Commit af664d5

Browse files
committed
Add URL shortner example
1 parent 1a2e99f commit af664d5

7 files changed

Lines changed: 168 additions & 0 deletions

File tree

README.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ Need to deploy your Worker to Cloudflare? Python Workers are in open beta and ha
3131
- [**`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.
3232
- [**`chatroom/`**](chatroom) - A real-time chatroom using WebSocket.
3333
- [**`sync-http-clients/`**](sync-http-clients) — demonstrates outbound HTTP with synchronous Python clients (`requests`, `urllib3`, and `httpx.Client`).
34+
- [**`url-shortener/`**](url-shortener) — a URL shortener backed by Workers KV.
3435
- [**`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.
3536
- [**`django/`**](django) — runs a naive Django WSGI application directly on Python Workers.
3637
- [**`django-todo-d1/`**](django-todo-d1) — implements the Todo-Backend API with Django and D1.

tests/test_examples.py

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,27 @@ def test_sync_http_clients(dev_server):
129129
assert result["saw_expected_text"] is True
130130

131131

132+
def test_url_shortener(dev_server):
133+
port = dev_server
134+
base = f"http://localhost:{port}"
135+
destination = "https://example.com/path"
136+
137+
response = requests.post(f"{base}/shorten", json={"url": destination})
138+
assert response.status_code == 201
139+
shortened = response.json()
140+
assert len(shortened["code"]) == 8
141+
assert shortened["short_url"] == f"{base}/{shortened['code']}"
142+
assert shortened["url"] == destination
143+
144+
response = requests.get(shortened["short_url"], allow_redirects=False)
145+
assert response.status_code == 302
146+
assert response.headers["location"] == destination
147+
148+
response = requests.get(f"{base}/missing-code")
149+
assert response.status_code == 404
150+
assert response.json()["error"] == "not found"
151+
152+
132153
def test_cron(dev_server):
133154
port = dev_server
134155
response = requests.get(f"http://localhost:{port}")

url-shortener/README.md

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
# URL Shortener Example
2+
3+
[![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)
4+
5+
A Python Worker that creates short URLs backed by Workers KV.
6+
7+
## Configure Workers KV
8+
9+
Create a namespace, then replace the placeholder ID in `wrangler.jsonc` with the ID from the command output:
10+
11+
```sh
12+
uv run pywrangler kv namespace create LINKS
13+
```
14+
15+
## Run locally
16+
17+
First ensure that [uv](https://docs.astral.sh/uv/getting-started/installation/#standalone-installer) is installed. Then run:
18+
19+
```sh
20+
uv run pywrangler dev
21+
curl -X POST http://localhost:8787/shorten \
22+
-H 'content-type: application/json' \
23+
-d '{"url":"https://example.com/path"}'
24+
```
25+
26+
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`.
27+
28+
## API
29+
30+
| Endpoint | Description |
31+
|---|---|
32+
| `POST /shorten` | Store an absolute HTTP(S) URL and return a short URL. |
33+
| `GET /<code>` | Redirect to the stored URL with status `302`. |

url-shortener/package.json

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
{
2+
"name": "python-url-shortener",
3+
"version": "0.0.0",
4+
"private": true,
5+
"scripts": {
6+
"deploy": "uv run pywrangler deploy",
7+
"dev": "uv run pywrangler dev",
8+
"start": "uv run pywrangler dev"
9+
},
10+
"devDependencies": {
11+
"wrangler": "^4.114.0"
12+
}
13+
}

url-shortener/pyproject.toml

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
[project]
2+
name = "python-url-shortener"
3+
version = "0.1.0"
4+
description = "Workers KV URL shortener in Python"
5+
readme = "README.md"
6+
requires-python = ">=3.12"
7+
dependencies = []
8+
9+
[dependency-groups]
10+
dev = [
11+
"workers-py",
12+
"workers-runtime-sdk"
13+
]

url-shortener/src/entry.py

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
import uuid
2+
from urllib.parse import urlsplit
3+
4+
from workers import Response, WorkerEntrypoint
5+
6+
MAX_CODE_ATTEMPTS = 5
7+
8+
9+
def error(message, status, allow=None):
10+
headers = {"Allow": allow} if allow else None
11+
return Response.json({"error": message}, status=status, headers=headers)
12+
13+
14+
def is_valid_destination(url):
15+
try:
16+
parsed = urlsplit(url)
17+
hostname = parsed.hostname
18+
_ = parsed.port
19+
except ValueError:
20+
return False
21+
return parsed.scheme in ("http", "https") and bool(hostname)
22+
23+
24+
class Default(WorkerEntrypoint):
25+
async def fetch(self, request):
26+
path = urlsplit(request.url).path
27+
28+
if path == "/shorten":
29+
if request.method != "POST":
30+
return error("method not allowed", 405, allow="POST")
31+
return await self.shorten(request)
32+
33+
if request.method != "GET":
34+
return error("method not allowed", 405, allow="GET")
35+
36+
code = path.lstrip("/")
37+
if not code or "/" in code:
38+
return error("not found", 404)
39+
40+
url = await self.env.LINKS.get(code)
41+
if url is None:
42+
return error("not found", 404)
43+
return Response.redirect(url, 302)
44+
45+
async def shorten(self, request):
46+
try:
47+
body = await request.json()
48+
url = body["url"]
49+
except (KeyError, TypeError, ValueError):
50+
return error("request body must contain a URL", 400)
51+
52+
if not isinstance(url, str) or not is_valid_destination(url):
53+
return error("url must be an absolute HTTP(S) URL", 400)
54+
55+
for _ in range(MAX_CODE_ATTEMPTS):
56+
code = uuid.uuid4().hex[:8]
57+
58+
if await self.env.LINKS.get(code) is not None:
59+
# Code already exists, try again
60+
continue
61+
62+
await self.env.LINKS.put(code, url)
63+
origin = urlsplit(request.url)
64+
short_url = f"{origin.scheme}://{origin.netloc}/{code}"
65+
return Response.json(
66+
{"code": code, "short_url": short_url, "url": url}, status=201
67+
)
68+
69+
return error("could not allocate a short code", 503)

url-shortener/wrangler.jsonc

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
{
2+
"$schema": "node_modules/wrangler/config-schema.json",
3+
"name": "python-url-shortener",
4+
"main": "src/entry.py",
5+
"compatibility_date": "2026-09-01",
6+
"compatibility_flags": [
7+
"python_workers"
8+
],
9+
"kv_namespaces": [
10+
{
11+
"binding": "LINKS",
12+
"id": "<YOUR_KV_NAMESPACE_ID>"
13+
}
14+
],
15+
"observability": {
16+
"enabled": true
17+
}
18+
}

0 commit comments

Comments
 (0)