Skip to content

Examples

Last updated View as MarkdownAgent setup

Cloudflare has a wide range of Python examples in the Workers Example gallery.

In addition to those examples, consider the following ones that illustrate Python-specific behavior.

Modules in your Worker

Let's say your Worker has the following structure:

├── src
│   ├── module.py
│   └── main.py
├── uv.lock
├── pyproject.toml
└── wrangler.toml

In order to import module.py in main.py, you would use the following import statement:

import module

In this case, the main module is set to src/main.py in the wrangler.toml file like so:

main = "src/main.py"

This means that the src directory does not need to be specified in the import statement.

Parse an incoming request URL

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

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # 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(greeting)


        if url.path == "/favicon.ico":
          return Response("")

        return Response("Hello world!")

Parse JSON from the incoming request

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        body = await request.json()  # returns a native Python dict
        name = body["name"]
        return Response("Hello, {name}".format(name=name))

Return a JSON response

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        data = {"greeting": "Hello, World!", "status": "ok"}
        return Response.json(data)

Read bundled asset files in your Worker

Let's say your Worker has the following structure:

├── src
│   ├── file.html
│   └── main.py
└── wrangler.jsonc

In order to read a file in your Worker, you would do the following:

from pathlib import Path
from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        html_file = Path(__file__).parent / "file.html"
        return Response(html_file.read_text(), headers={"Content-Type": "text/html"})

Emit logs from your Python Worker

# To use the JavaScript console APIs
from js import console
from workers import WorkerEntrypoint, Response
# To use the native Python logging
import logging

class Default(WorkerEntrypoint):
    async def fetch(self, 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("We're testing logging!")

Publish to a Queue

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
			  # Bindings are available on the 'env' attribute
        # https://developers.cloudflare.com/queues/

        # The default contentType is "json"