Python Workers

You can use Hyperdrive with Python Workers. To achieve this, you need to set your compatibility date to 2026-09-08 or later.

Supported drivers

Hyperdrive in Python Workers uses TCP socket support to establish database connections. While you can use any Python driver that works with TCP connections, we strongly recommend using the drivers in the tables below, as they have been tested and verified to work with Hyperdrive.

PostgreSQL

Driver	Documentation
`asyncpg` (recommended)	asyncpg documentation ↗
`pg8000`	pg8000 documentation ↗
`psycopg`	psycopg documentation ↗

MySQL

Driver	Documentation
`aiomysql` (recommended)	aiomysql documentation ↗
`pymysql`	pymysql documentation ↗

Connect to your database

Before you begin, create a Python Worker and create a Hyperdrive configuration for your database.

Add the Hyperdrive binding to your Wrangler configuration. Replace <HYPERDRIVE_CONFIG_ID> with your configuration 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-17",
  "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-17"
compatibility_flags = ["python_workers"]

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

Install your driver and replace src/main.py with the corresponding example.

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

Deploy your Worker:
```
uv run pywrangler deploy
```

Limitations and compatibility

Socket support

TCP socket support in Python Workers internally uses the connect API. While most standard library socket operations are supported, some low-level operations might not work as expected.

Concurrency and async safety

Socket operations in Python Workers do not block the event loop. Although Python's native socket operations are synchronous, the underlying TCP socket implementation in Python Workers is asynchronous. This allows multiple requests to be processed concurrently while one request waits for a socket operation to complete.

To ensure synchronous database operations are serialized, use a lock to prevent concurrent access:

import asyncio

lock = asyncio.Lock()

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

SQLAlchemy support

Currently, only synchronous SQLAlchemy ORMs are supported in Python Workers. Async SQLAlchemy ORMs are not yet supported due to a lack of greenlet support in the Python Workers environment.