Python Workers

Python Workers で Hyperdrive を使えます。 Python Workers で Hyperdrive を使うには、互換日を 2026-09-08 以降に設定します。

対応ドライバー

Python Workers の Hyperdrive は、データベース接続の確立に TCP ソケットサポートを使います。 TCP 接続で動く任意の Python ドライバーを使えますが、Hyperdrive で動作確認済みのため、次の表のドライバーを強く推奨します。

PostgreSQL

ドライバー	ドキュメント
`asyncpg`（推奨）	asyncpg ドキュメント ↗
`pg8000`	pg8000 ドキュメント ↗
`psycopg`	psycopg ドキュメント ↗

MySQL

ドライバー	ドキュメント
`aiomysql`（推奨）	aiomysql ドキュメント ↗
`pymysql`	pymysql ドキュメント ↗

データベースに接続する

始める前に、Python Worker を作成し、データベース向けに Hyperdrive 設定を作成します。

Wrangler 設定に Hyperdrive バインディングを追加します。<HYPERDRIVE_CONFIG_ID> は設定 ID に置き換えます。

{
  "$schema": "./node_modules/wrangler/config-schema.json",
  "name": "python-hyperdrive",
  "main": "src/main.py",
  // Set this to today's date
  "compatibility_date": "2026-09-20",
  "compatibility_flags": [
    "python_workers"
  ],
  "hyperdrive": [
    {
      "binding": "HYPERDRIVE",
      "id": "<HYPERDRIVE_CONFIG_ID>"
    }
  ]
}

name = "python-hyperdrive"
main = "src/main.py"
# Set this to today's date
compatibility_date = "2026-09-20"
compatibility_flags = ["python_workers"]

[[hyperdrive]]
binding = "HYPERDRIVE"
id = "<HYPERDRIVE_CONFIG_ID>"

ドライバーをインストールし、src/main.py を対応する例に置き換えます。

[project]
dependencies = [
    "asyncpg",
]

src/main.pypython

from contextlib import closing

import asyncpg
from workers import Response, WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        hd = self.env.HYPERDRIVE
        connection = await asyncpg.connect(
            host=hd.host,
            port=int(hd.port),
            user=hd.user,
            password=hd.password,
            database=hd.database,
            ssl=False,
        )
        await connection.execute("SELECT 1")
        await connection.close()

[project]
dependencies = [
    "aiomysql",
]

src/main.pypython

import aiomysql
from workers import Response, WorkerEntrypoint


class Default(WorkerEntrypoint):
    async def fetch(self, request):
        hd = self.env.HYPERDRIVE
        connection = await aiomysql.connect(
            host=hd.host,
            port=int(hd.port),
            user=hd.user,
            password=hd.password,
            db=hd.database,
            ssl=None,
        )
        try:
            cursor = await connection.cursor()
            await cursor.execute("SELECT 1")
            result = await cursor.fetchone()
            return Response.json({"result": result[0]})
        finally:
            connection.close()

Worker をデプロイします。
```
uv run pywrangler deploy
```

制限と互換性

ソケットサポート

Python Workers の TCP ソケットサポートは、内部で connect API を使います。標準ライブラリのソケット操作のほとんどはサポートされますが、一部の低レベル操作は想定どおりに動かないことがあります。

並行性と非同期安全性

Python Workers のソケット操作はイベントループをブロックしません。 Python のネイティブソケット操作は同期ですが、Python Workers の基になる TCP ソケット実装は非同期です。これにより、1 つのリクエストがソケット操作の完了を待っているあいだも、複数のリクエストを並行して処理できます。

同期のデータベース操作を直列化するには、ロックを使い、同時アクセスを防ぎます。

import asyncio

lock = asyncio.Lock()

async with lock:
    # Your database operation here
    synchronous_db_operation()

SQLAlchemy サポート

現在、Python Workers でサポートされるのは同期 SQLAlchemy ORM のみです。 Python Workers 環境に greenlet サポートがないため、非同期 SQLAlchemy ORM はまだサポートされていません。