Turbopuffer Python Client Library

In API responses, a field might be explicitly null or missing entirely. The library represents both as None. To distinguish between them, check if the field name exists in the .model_fields_set attribute of the response object.

if response.my_field is None:
  if 'my_field' not in response.model_fields_set:
    print('Got json like {}, without a "my_field" key present at all.')
  else:
    print('Got json like {"my_field": null}.')

The library provides auto-paginating iterators for list methods. You can iterate directly over the response object in a for loop (sync) or async for loop (async).

# Synchronous auto-pagination
for ns in client.namespaces(prefix="products"):
    print(ns.id)

# Asynchronous auto-pagination
async for ns in client.namespaces(prefix="products"):
    print(ns.id)

Starting from v0.5, the client is initialized using the Turbopuffer class. This allows you to manage multiple regions by creating different client instances. It is recommended to use the region parameter instead of base_url where possible.

import turbopuffer
import os

tpuf = turbopuffer.Turbopuffer(
    api_key=os.getenv('TURBOPUFFER_API_KEY'),
    region='gcp-us-central1',
)

# You can also create clients for other regions
tpuf_eu = turbopuffer.Turbopuffer(region='gcp-eu-west1')

In version 2.0, the copy_from_namespace and branch_from_namespace parameters were removed from the write method and replaced with dedicated methods: .copy_from() and .branch_from().

# New v2.0 syntax for copying
tpuf.namespace("ns").copy_from(
    source_namespace="src",
    source_region="gcp-us-central1",
)

# New v2.0 syntax for branching
tpuf.namespace("ns").branch_from(source_namespace="src")

By default, the async client uses httpx. To use aiohttp for improved concurrency performance, install the aiohttp extra and pass DefaultAioHttpClient() to the client.

# install from PyPI
pip install turbopuffer[aiohttp]

import os
import asyncio
from turbopuffer import DefaultAioHttpClient
from turbopuffer import AsyncTurbopuffer

async def main() -> None:
    async with AsyncTurbopuffer(
        api_key=os.environ.get("TURBOPUFFER_API_KEY"),
        http_client=DefaultAioHttpClient(),
    ) as client:
        # ... use client
        pass

asyncio.run(main())

If you need to access undocumented endpoints, parameters, or response properties, you can use the following methods:

• Undocumented Endpoints: Use HTTP verb methods like client.get, client.post, etc. Use the cast_to parameter to specify the expected return type (e.g., httpx.Response).
• Undocumented Request Params: Use extra_query, extra_body, or extra_headers to send additional data.
• Undocumented Response Properties: Access extra fields directly via response.unknown_prop or retrieve all extra fields as a dictionary using response.model_extra.

import httpx

# Undocumented endpoint
response = client.post(
    "/foo",
    cast_to=httpx.Response,
    body={"my_param": True},
)

print(response.headers.get("x-foo"))

You can use the synchronous Turbopuffer client or the asynchronous AsyncTurbopuffer client. It is recommended to provide your API key via the TURBOPUFFER_API_KEY environment variable. You can also specify a region (e.g., gcp-us-central1).

import os
from turbopuffer import Turbopuffer

tpuf = Turbopuffer(
    region="gcp-us-central1",
    api_key=os.environ.get("TURBOPUFFER_API_KEY"),
)

While the library automatically closes connections when the client is garbage collected, it is best practice to manually manage the lifecycle using a context manager or by calling .close() to ensure resources are released promptly.

from turbopuffer import Turbopuffer

with Turbopuffer() as client:
  # make requests here
  ...

# HTTP client is now closed

In version 2.0, the encryption parameter structure changed. Instead of a nested cmek key, use a flat dictionary with a mode and key_name.

To migrate a namespace from Customer Managed Encryption Keys (CMEK) back to default encryption, use {"mode": "default"}.

# New v2.0 syntax
tpuf.namespace("ns").write(
    upsert_rows=[...],
    encryption={"mode": "customer-managed", "key_name": "..."},
)

For asynchronous applications, import AsyncTurbopuffer and await the API calls. The functionality is identical to the synchronous client.

import os
import asyncio
from turbopuffer import AsyncTurbopuffer

tpuf = AsyncTurbopuffer(
    region="gcp-us-central1",
    api_key=os.environ.get("TURBOPUFFER_API_KEY"),
)

ns = tpuf.namespace("example")

async def main() -> None:
    vector_result = await ns.query(
        rank_by=("vector", "ANN", [0.1, 0.2]),
        top_k=10,
    )
    print(vector_result.rows)

asyncio.run(main())

Install the library from PyPI using pip.

# install from PyPI
pip install turbopuffer

The library uses the standard Python logging module. You can control the verbosity of logs by setting the TURBOPUFFER_LOG environment variable.

Set it to info for standard logging or debug for more verbose output.

$ export TURBOPUFFER_LOG=info