跳转到内容
搜索文档

从 Python Workers 查询 D1

了解如何从 Python Worker 查询 D1

最后更新 查看 MarkdownAgent 设置

Cloudflare Workers 平台支持多种语言,包括 TypeScript、JavaScript、Rust 和 Python。本指南展示如何从 Python 查询 D1 数据库并全球部署您的应用。

前提条件

开始之前,您应该:

  1. 阅读 D1 教程(TypeScript 和 JavaScript 版本),了解如何创建 D1 数据库并配置 Workers 项目
  2. 参阅 Python 语言指南,了解 Workers 平台上的 Python 支持。
  3. 基本熟悉 Python 语言。

如果您是 Cloudflare Workers 新用户,请先参阅快速入门指南,然后再继续本示例。

从 Python 查询

本示例假设您已有 D1 数据库。要让 Python Worker 查询数据库,首先需要在 Worker 和 D1 数据库之间创建绑定(binding),并在 Wrangler 配置文件中定义。

您需要 D1 数据库的 database_namedatabase_id。您可以使用 wrangler CLI 创建新数据库或获取现有数据库的 ID,如下所示:

Create a databasesh
npx wrangler d1 create my-first-db
Retrieve a database IDsh
npx wrangler d1 info some-existing-db
# ┌───────────────────┬──────────────────────────────────────┐
# │                   │ c89db32e-83f4-4e62-8cd7-7c8f97659029 │
# ├───────────────────┼──────────────────────────────────────┤
# │ name              │ db-enam                              │
# ├───────────────────┼──────────────────────────────────────┤
# │ created_at        │ 2023-06-12T16:52:03.071Z             │
# └───────────────────┴──────────────────────────────────────┘

1. 配置绑定

在 Wrangler 文件中,创建新的 [[d1_databases]] 配置块,将 database_namedatabase_id 设置为您要查询的 D1 数据库的名称和 ID:

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "python-and-d1",
	"main": "src/entry.py",
	"compatibility_flags": [ // Required for Python Workers
		"python_workers"
	],
	// Set this to today's date
	"compatibility_date": "2026-08-17",
	"d1_databases": [
		{
			"binding": "DB", // This will be how you refer to your database in your Worker
			"database_name": "YOUR_DATABASE_NAME",
			"database_id": "YOUR_DATABASE_ID"
		}
	]
}
"$schema" = "./node_modules/wrangler/config-schema.json"
name = "python-and-d1"
main = "src/entry.py"
compatibility_flags = [ "python_workers" ]
# Set this to today's date
compatibility_date = "2026-08-17"

[[d1_databases]]
binding = "DB"
database_name = "YOUR_DATABASE_NAME"
database_id = "YOUR_DATABASE_ID"

binding 的值是您在 Worker 中引用数据库的方式。如果更改此值,也必须在 Worker 脚本中更改。

2. 创建 Python Worker

要创建 Python Worker,在 src/entry.py 创建一个空文件,与 Wrangler 文件中 main 的值匹配,内容如下:

from workers import Response, WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        # Do anything else you'd like on request here!

        try:
            # Query D1 - we'll list all tables in our database in this example
            results = await self.env.DB.prepare("PRAGMA table_list").run()
            # Return a JSON response
            return Response.json(results)
        except Exception as e:
            return Response.json({"error": "Database query failed"}, status=500)

Wrangler 文件中 binding 的值必须与 Python 代码中的变量名完全匹配。本示例通过 DB 绑定引用数据库,并通过 await self.env.DB.prepare(...) 查询此绑定。

然后您可以直接部署 Python Worker:

npx wrangler deploy
# Example output
#
# Your worker has access to the following bindings:
# - D1 Databases:
#   - DB: db-enam (c89db32e-83f4-4e62-8cd7-7c8f97659029)
# Total Upload: 0.18 KiB / gzip: 0.17 KiB
# Uploaded python-and-d1 (4.93 sec)
# Published python-and-d1 (0.51 sec)
#   https://python-and-d1.YOUR_SUBDOMAIN.workers.dev
# Current Deployment ID: 80b72e19-da82-4465-83a2-c12fb11ccc72

您的 Worker 将在 https://python-and-d1.YOUR_SUBDOMAIN.workers.dev 可用。

如果部署时收到错误:

后续步骤

这篇文档对您有帮助吗?