运行 Python 代码

在 sandbox 中运行 Python 代码并返回其输出。镜像决定 Python 版本以及代码可以导入的软件包。一次运行写入的文件会留在 sandbox 中供下一次运行使用。

前提条件

带有 Durable Object 的 Worker，该 Durable Object 使用 Durable Object 调度策略启动 container。要创建一个，请参阅运行 Linux 命令。

当您运行 wrangler deploy 时，必须在本地运行 Docker。对于大多数人来说，安装 Docker 的最佳方式是按照安装 Docker Desktop 的文档 ↗进行操作。其他工具（如 Colima ↗）也可能有效。

您可以通过在终端中运行 docker info 命令来检查 Docker 是否正常运行。如果 Docker 正在运行，该命令将成功执行。如果 Docker 未运行，docker info 命令将挂起或返回包含 "Cannot connect to the Docker daemon" 消息的错误。

运行 Python 代码

创建名为 sales.py 的文件，写入要运行的 Python 代码。本例使用 pandas：

sales.pypython

import pandas as pd

sales = pd.DataFrame({"city": ["Lisbon", "Austin", "Lisbon"], "units": [3, 5, 4]})
print(sales.groupby("city")["units"].sum())

在项目根目录创建 Dockerfile，包含 Python 以及代码可以导入的软件包：
Dockerfiledockerfile
```
FROM python:3.13-slim

RUN pip install --no-cache-dir numpy pandas matplotlib

WORKDIR /workspace

CMD ["sleep", "infinity"]
```
在镜像中安装软件包后，sandbox 启动即可使用它们，无需 sandbox 访问互联网。
在 wrangler.jsonc 中，将 Dockerfile 构建为 containers 条目中的具名镜像，然后生成类型：
{ "containers": [ { "class_name": "MyContainer", "scheduling_policy": "durable_object", "images": { "python": { "dockerfile": "./Dockerfile", }, }, }, ], }
[[containers]] class_name = "MyContainer" scheduling_policy = "durable_object" [containers.images.python] dockerfile = "./Dockerfile"
npx wrangler types
yarn wrangler types
pnpm wrangler types

为 Durable Object 添加不活动超时。当重启后的 Durable Object 发现 container 仍在运行时，构造函数会再次设置超时：

src/index.tsts

const INACTIVITY_TIMEOUT_MS = 10 * 60 * 1000;

export class MyContainer extends DurableObject<Env> {
    constructor(ctx: DurableObjectState, env: Env) {
        super(ctx, env);
        const container = ctx.container;
        if (container?.running) {
            void ctx.blockConcurrencyWhile(() =>
                container.setInactivityTimeout(INACTIVITY_TIMEOUT_MS),
            );
        }
    }
}

然后添加运行 Python 代码的方法：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    // ...

    async runPython(code: string) {
        const container = this.ctx.container;

        if (!container) {
            throw new Error("No container is configured");
        }

        if (!container.running) {
            container.start({
                image: container.images.python,
                // Block the code from reaching the Internet
                enableInternet: false,
            });
            await container.setInactivityTimeout(INACTIVITY_TIMEOUT_MS);
        }

        // Each run is a new Python process, so variables do not carry over
        const process = await container.exec(
            [
                // GNU coreutils `timeout` stops code that runs longer than 60 seconds
                "timeout", "--kill-after=5", "60",
                // `python3 -` reads the code from standard input
                "python3", "-",
            ],
            {
                stdin: new Response(code).body!,
                // Files that the code writes here stay until the instance stops
                cwd: "/workspace",
            },
        );
        const output = await process.output();
        const decoder = new TextDecoder();

        return {
            stdout: decoder.decode(output.stdout),
            // A Python exception prints its traceback here
            stderr: decoder.decode(output.stderr),
            // `124` or `137` if the code timed out, `1` if it raised an exception
            exitCode: output.exitCode,
        };
    }
}

关于 timeout 如何停止代码启动的进程，请参阅停止命令启动的进程。

output() 会将代码打印的全部内容保存在 Durable Object 的内存中。对于输出很多的代码，请改为流式输出。

向 Worker 添加路由，将请求正文作为 Python 代码运行：

src/index.jsjs

export default {
	async fetch(request, env) {
		const url = new URL(request.url);
		const match =
			/^\/sandboxes\/([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)\/python$/.exec(
				url.pathname,
			);

		if (!match || request.method !== "POST") {
			return new Response("Not found", { status: 404 });
		}

		const sandbox = env.MY_CONTAINER.getByName(match[1]);
		return Response.json(await sandbox.runPython(await request.text()));
	},
};

src/index.tsts

export default {
    async fetch(request: Request, env: Env): Promise<Response> {
        const url = new URL(request.url);
        const match = /^\/sandboxes\/([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)\/python$/.exec(url.pathname);

        if (!match || request.method !== "POST") {
            return new Response("Not found", { status: 404 });
        }

        const sandbox = env.MY_CONTAINER.getByName(match[1]);
        return Response.json(await sandbox.runPython(await request.text()));
    },
} satisfies ExportedHandler<Env>;

代码可以读取其 sandbox 中的每个文件，因此请为每个用户提供独立的 sandbox 名称，并先认证调用方。更多信息请参阅 Sandbox 安全。

部署你的 Worker：
npx wrangler deploy
yarn wrangler deploy
pnpm wrangler deploy

在名为 ada 的 sandbox 中运行 sales.py。将示例 hostname 替换为 Wrangler 打印的 workers.dev URL：

curl https://<YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev/sandboxes/ada/python --data-binary @sales.py

{
    "stdout": "city\nAustin    5\nLisbon    7\nName: units, dtype: int64\n",
    "stderr": "",
    "exitCode": 0
}

返回图表

要返回图像（例如 matplotlib 图表），请让代码将其保存到 /workspace，然后在另一次请求中读取该文件：

向 Durable Object 添加从 /workspace 读取文件的方法：

src/index.tsts

export class MyContainer extends DurableObject<Env> {
    // ...

    async readFile(name: string) {
        const container = this.ctx.container;

        if (!container?.running) {
            return null;
        }

        const process = await container.exec(["cat", "--", name], {
            cwd: "/workspace",
            stderr: "ignore",
        });
        const output = await process.output();
        return output.exitCode === 0 ? output.stdout : null;
    }
}

在 Worker 的 fetch() handler 中，在解析 url 的那一行之后，添加返回图表的路由：

src/index.tsts

const chart = /^\/sandboxes\/([a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)\/chart\.png$/.exec(url.pathname);

if (chart) {
    const png = await env.MY_CONTAINER.getByName(chart[1]).readFile("chart.png");
    return png
        ? new Response(png, { headers: { "Content-Type": "image/png" } })
        : new Response("Not found", { status: 404 });
}

运行保存图表的代码，然后在浏览器中打开 /sandboxes/ada/chart.png：

chart.pypython

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt

plt.bar(["Austin", "Lisbon"], [5, 7])
plt.savefig("chart.png")

curl https://<YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev/sandboxes/ada/python --data-binary @chart.py

matplotlib.use("Agg") 会在没有显示器的情况下绘图。对于更大的文件和上传，请参阅在 sandbox 内外移动文件。

运行其他语言

要运行其他语言，请更改基础镜像以及从标准输入读取的命令。例如，使用 FROM ruby:3.4-slim 和 ["ruby", "-"]，或 FROM node:24-slim 和 ["node", "-"]。

运行 Python 代码

前提条件

运行 Python 代码

返回图表

运行其他语言

相关资源