Examples
Return a custom status code and/or response headers
src/entry.pyfrom js import Response, Headers
async def on_fetch(request, env): # Create a Headers object headers = Headers.new({"x-hello-from": "python-workers"}.items()) # Return a response object with a status code and headers return Response.new("Hello world!", status=404, headers=headers)
Parse an incoming request URL
src/entry.pyfrom js import Responsefrom urllib.parse import urlparse, parse_qs
async def on_fetch(request, env): # Parse the incoming request URL url = urlparse(request.url) # Parse the query parameters into a Python dictionary params = parse_qs(url.query)
if "name" in params: greeting = "Hello there, {name}".format(name=params["name"][0]) return Response.new(greeting)
if url.path == "/favicon.ico": return Response.new("")
return Response.new("Hello world!")
Parse JSON from the incoming request
src/entry.pyfrom js import Response
async def on_fetch(request): name = (await request.json()).name return Response.new("Hello, {name}".format(name=name))
Emit logs from your Python Worker
src/entry.py# To use the JavaScript console APIs
from js import console, Response# To use the native Python logging
import logging
async def on_fetch(request): # Use the console APIs from JavaScript # https://developer.mozilla.org/en-US/docs/Web/API/console console.log("console.log from Python!")
# Alternatively, use the native Python logger logger = logging.getLogger(__name__)
# The default level is warning. We can change that to info. logging.basicConfig(level=logging.INFO)
logger.error("error from Python!") logger.info("info log from Python!")
# Or just use print() print("print() from Python!")
return Response.new("We're testing logging!")
Respond with JSON
src/entry.pyfrom js import Response, Headersimport json
async def on_fetch(request): # Use json.loads to serialize Python objects to JSON strings payload = json.dumps({"c": 0, "b": 0, "a": 0}, sort_keys=True)
headers = Headers.new({"content-type": "application/json"}.items()) return Response.new(payload, headers=headers)
Publish to a Queue
src/entry.pyfrom js import Response, Objectfrom pyodide.ffi import to_js as _to_js
# to_js converts between Python dictionaries and JavaScript Objects
def to_js(obj): return _to_js(obj, dict_converter=Object.fromEntries)
async def on_fetch(request, env): # Bindings are available on the 'env' parameter # https://developers.cloudflare.com/queues/
# The default contentType is "json" # We can also pass plain text strings await env.QUEUE.send("hello", contentType="text") # Send a JSON payload await env.QUEUE.send(to_js({"hello": "world"}))
# Return a response return Response.json(to_js({"write": "success"}))
Query a D1 Database
src/entry.pyfrom js import Response
async def on_fetch(request, env): results = await env.DB.prepare("PRAGMA table_list").all() # Return a JSON response return Response.json(results)
Refer to Query D1 from Python Workers for a more in-depth tutorial that covers how to create a new D1 database and configure bindings to D1.
Next steps
- If you’re new to Workers and Python, refer to the get started guide
- Learn more about calling JavaScript methods and accessing JavaScript objects from Python
- Understand the supported packages and versions currently available to Python Workers.