← Cloudflare Workers / workers / languages / python

Примеры

У Cloudflare есть широкий набор примеров на Python в Галерея примеров Workers.

Помимо этих примеров, рассмотрите следующие, которые иллюстрируют особенности поведения, специфичные для Python.

Модули в вашем Worker

Предположим, что ваш Worker имеет следующую структуру:

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

Чтобы импортировать module.py в main.py, вы бы использовали следующую инструкцию импорта:

import module

В этом случае в качестве основного модуля указан src/main.py в файле wrangler.toml следующим образом:

main = "src/main.py"

Это означает, что src каталог не нужно указывать в инструкции import.

Разобрать 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!")

Разобрать JSON из входящего запроса

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))

Возвращает JSON-ответ

from workers import WorkerEntrypoint, Response

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

Чтение упакованных файлов ресурсов в Worker

Предположим, что ваш Worker имеет следующую структуру:

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

Чтобы прочитать файл в Worker, выполните следующее:

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"})

Отправка логов из вашего 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!")

Публикация в 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"
        # We can also pass plain text strings
        await self.env.QUEUE.send("hello", contentType="text")
        # Send a JSON payload
        await self.env.QUEUE.send({"hello": "world"})

        return Response.json({"write": "success"})

Запрос базы данных D1

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        results = await self.env.DB.prepare("PRAGMA table_list").run()
        # Return a JSON response
        return Response.json(results)

См. Запрос D1 из Python Workers более подробное руководство о том, как создать новую базу данных D1 и настроить привязки к D1.

Durable Object

from workers import WorkerEntrypoint, Response, DurableObject

class List(DurableObject):
    async def get_messages(self):
        messages = await self.ctx.storage.get("messages")
        return messages if messages else []

    async def add_message(self, message):
        messages = await self.get_messages()
        messages.append(message)
        await self.ctx.storage.put("messages", messages)
        return

    async def say_hello(self):
        result = self.ctx.storage.sql.exec(
            "SELECT 'Hello, World!' as greeting"
        ).one()

        return result.greeting

См. Документация по Durable Objects, где это описано подробнее.

Cron Trigger

from workers import WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def scheduled(self, controller, env, ctx):
        # All four parameters (self, controller, env, ctx) are required —
        # unlike fetch() which only takes (self, request).
        print("cron processed")

См. Документация по Cron Triggers, где это описано подробнее.

Workflows

from workers import WorkflowEntrypoint

class MyWorkflow(WorkflowEntrypoint):
    async def run(self, event, step):
        @step.do()
        async def step_a():
            # do some work
            return 10

        @step.do()
        async def step_b():
            # do some work
            return 20

        @step.do(concurrent=True)
        async def my_final_step(step_a, step_b):
            # should return 30
            return step_a + step_b

        await my_final_step()

См. Документация Python Workflows, где это описано подробнее.

Больше примеров

Или можно клонировать репозиторий с примерами ↗ чтобы посмотреть ещё больше примеров:

git clone https://github.com/cloudflare/python-workers-examples