bubble-data-api-client 0.8.0

Query your Bubble.io database from Python with an async client and Pydantic ORM. Type-safe CRUD for the Bubble Data API with connection pooling and retries.

bubble-data-api-client

Query your Bubble.io database from Python. A fast, async client and Pydantic ORM for the Bubble Data API, with type-safe CRUD, connection pooling, and configurable retries.

Why use this?

If you're integrating Python with a Bubble app, this library handles the boilerplate so you can focus on your logic.

Common use cases:

• Syncing data between Bubble and external systems
• Data migrations and bulk imports
• Backend scripts and automation
• Reporting that pulls from Bubble's database

Clean, simple interface

# create
user = await User.create(name="Ada", email="[e-mail]")

# retrieve
user = await User.get(uid)

# query (paginated)
users = await User.find(constraints=[
    constraint("status", ConstraintType.EQUALS, "active")
])

# query all matching records
all_users = await User.find_all()

# iterate with constant memory
async for user in User.find_iter():
    process(user)

# update
user.name = "Ada Lovelace"
await user.save()

# delete
await user.delete()

# check existence
if await User.exists(uid):
    print("User exists")

# count
active_count = await User.count(constraints=[
    constraint("status", ConstraintType.EQUALS, "active")
])

IDE support and type checking

Models provide autocomplete and catch errors before runtime:

class User(BubbleModel, typename="user"):
    name: str
    email: str
    age: int

user = await User.get(uid)
user.name    # IDE autocomplete works
user.nme     # Typo caught by pyright/mypy

Works with pyright, mypy, and IDE type checkers.

Validation catches bad data early

Pydantic validates data when models are created:

# Type mismatch caught immediately
user = User(_id="123x456", name="Ada", email="[e-mail]", age="twenty-five")
# ValidationError: Input should be a valid integer

# Invalid Bubble UID caught at the model level
class Order(BubbleModel, typename="order"):
    customer: BubbleUID

order = Order(_id="123x456", customer="not-a-valid-uid")
# ValidationError: invalid Bubble UID format: not-a-valid-uid

Bubble-specific handling

The library handles Bubble's API quirks automatically:

• Field mapping: Bubble's _id field maps to uid on your models
• Response parsing: Extracts data from Bubble's nested {"response": {"results": [...]}} structure
• Constraint format: Builds the JSON constraint format Bubble expects

Duplicate handling

Bubble doesn't enforce unique constraints, so duplicates can occur. The create_or_update method provides strategies to handle this:

# if duplicates exist, keep the oldest (by created date) and delete the rest
user, created = await User.create_or_update(
    match={"external_id": "ext-123"},
    create_data={"name": "Canonical Name"},
    update_data={"name": "Canonical Name"},
    on_multiple=OnMultiple.DEDUPE_OLDEST_CREATED,
)

Connection reuse

HTTP connections are pooled per event loop, avoiding reconnection overhead when making multiple requests

Features

• Async-first: built on httpx with HTTP/2
• Pydantic ORM: define models once, get validation and autocomplete
• Connection pooling: automatic per-event-loop client reuse
• Rich query constraints: pythonic filtering using Bubble's constraint system
• Efficient iteration: find_iter() streams records with constant memory
• Upsert with duplicate handling: create_or_update with configurable strategies
• Configurable retries: plug in your own retry policy via tenacity
• UID validation: catch invalid Bubble IDs at the model level

Installation

pip install bubble-data-api-client

Requires Python 3.12+.

Quick Start

Configuration

from bubble_data_api_client import configure

configure(
    data_api_root_url="https://your-app.bubbleapps.io/api/1.1/obj",
    api_key="your-api-key",
)

Or use a dynamic provider for secrets management:

import os
from bubble_data_api_client import set_config_provider, BubbleConfig

def get_config() -> BubbleConfig:
    return BubbleConfig(
        data_api_root_url=os.environ["BUBBLE_API_URL"],
        api_key=os.environ["BUBBLE_API_KEY"],
    )

set_config_provider(get_config)

Using the ORM

Define typed models with validation:

from bubble_data_api_client import BubbleModel, BubbleUID, OptionalBubbleUID

class User(BubbleModel, typename="user"):
    name: str
    email: str
    company: OptionalBubbleUID = None  # linked Bubble record

class Company(BubbleModel, typename="company"):
    name: str
    industry: str

Then use them:

# create
user = await User.create(name="Ada Lovelace", email="[e-mail]")

# retrieve
user = await User.get("1234567890x1234567890")

# query with constraints (single page)
from bubble_data_api_client import constraint, ConstraintType

active_users = await User.find(constraints=[
    constraint("status", ConstraintType.EQUALS, "active"),
    constraint("age", ConstraintType.GREATER_THAN, 18),
])

# get all matching records as a list
all_active = await User.find_all(constraints=[
    constraint("status", ConstraintType.EQUALS, "active"),
])

# iterate through all records with constant memory
async for user in User.find_iter():
    print(user.name)

# update
user.name = "Ada L."
await user.save()

# delete
await user.delete()

Smart Upserts

The create_or_update method handles the common "upsert" pattern with configurable strategies for handling duplicates:

from bubble_data_api_client import OnMultiple

# basic upsert, matches by external_id and creates if not found
user, created = await User.create_or_update(
    match={"external_id": "ext-123"},
    create_data={"name": "New User", "email": "[e-mail]"},
    update_data={"email": "[e-mail]"},
    on_multiple=OnMultiple.ERROR,
)
# returns (User, bool): the instance and whether it was created

match fields locate the record. create_data is merged with match when inserting a new record; update_data is applied when a record already exists. At least one of create_data or update_data must be provided.

Duplicate Handling Strategies

Since Bubble doesn't enforce unique constraints, duplicates can occur. Choose how to handle them:

Strategy	Behavior
OnMultiple.ERROR	Raise MultipleMatchesError (fail-fast)
OnMultiple.UPDATE_FIRST	Update first match (arbitrary order)
OnMultiple.UPDATE_ALL	Update all matches concurrently
OnMultiple.DEDUPE_OLDEST_CREATED	Keep oldest by Created Date, delete others, then update
OnMultiple.DEDUPE_NEWEST_CREATED	Keep newest by Created Date, delete others, then update
OnMultiple.DEDUPE_OLDEST_MODIFIED	Keep oldest by Modified Date, delete others, then update
OnMultiple.DEDUPE_NEWEST_MODIFIED	Keep newest by Modified Date, delete others, then update

# auto-deduplicate, keeping the oldest record by Created Date
user, created = await User.create_or_update(
    match={"external_id": "ext-123"},
    create_data={"name": "Canonical Name"},
    update_data={"name": "Canonical Name"},
    on_multiple=OnMultiple.DEDUPE_OLDEST_CREATED,
)

Constraints

Build type-safe queries using Bubble's constraint system:

from bubble_data_api_client import constraint, ConstraintType

constraints = [
    constraint("status", ConstraintType.EQUALS, "active"),
    constraint("age", ConstraintType.GREATER_THAN, 21),
    constraint("tags", ConstraintType.CONTAINS, "premium"),
    constraint("email", ConstraintType.IS_NOT_EMPTY),
    constraint("category", ConstraintType.IN, ["A", "B", "C"]),
]

results = await User.find(constraints=constraints)

Available constraint types: EQUALS, NOT_EQUAL, IS_EMPTY (any field), IS_NOT_EMPTY (any field), TEXT_CONTAINS, NOT_TEXT_CONTAINS, GREATER_THAN, LESS_THAN, IN, NOT_IN, CONTAINS, NOT_CONTAINS, EMPTY (list fields), NOT_EMPTY (list fields), GEOGRAPHIC_SEARCH.

Querying Records

Three methods for fetching records, depending on your needs:

Method	Returns	Use case
find()	list	Single page with manual pagination via cursor/limit
find_all()	list	All matching records collected into memory
find_iter()	AsyncIterator	All matching records with constant memory

# find(): single page, you control pagination
page1 = await User.find(limit=100)
page2 = await User.find(limit=100, cursor=100)

# find_all(): fetches all pages, returns when complete
all_users = await User.find_all(constraints=[...])
print(f"Got {len(all_users)} users")

# find_iter(): streams records with constant memory
async for user in User.find_iter(constraints=[...]):
    await process(user)  # each record processed as it arrives

Both find_all() and find_iter() handle pagination internally, fetching pages of page_size (default 100) until all records are retrieved.

Type-Safe Bubble UIDs

Validate Bubble record IDs at the type level:

from bubble_data_api_client import BubbleModel, BubbleUID, OptionalBubbleUID, OptionalBubbleUIDs

class Order(BubbleModel, typename="order"):
    customer: BubbleUID                    # required, validated
    referrer: OptionalBubbleUID = None     # optional, coerces invalid to None
    items: OptionalBubbleUIDs = None       # list of UIDs, filters invalid

# validation helpers
from bubble_data_api_client import is_bubble_uid, filter_bubble_uids

is_bubble_uid("1234567890x1234567890")  # True
is_bubble_uid("invalid")                 # False

filter_bubble_uids(["1661531100253x688916634279608300", "invalid", None])  # ["1661531100253x688916634279608300"]

Connection Pooling

Clients are automatically pooled per event loop. For explicit lifecycle control:

from bubble_data_api_client import client_scope, close_clients

# option 1: context manager (auto-closes on exit)
async with client_scope():
    await User.create(name="Test", email="[e-mail]")

# option 2: manual cleanup
await close_clients()

Retry Configuration

Plug in custom retry policies using tenacity:

import httpx
import tenacity
from bubble_data_api_client import configure

retry_policy = tenacity.AsyncRetrying(
    wait=tenacity.wait_exponential(multiplier=1, min=1, max=10),
    stop=tenacity.stop_after_attempt(3),
    retry=tenacity.retry_if_exception_type(httpx.TimeoutException),
)

configure(
    data_api_root_url="https://your-app.bubbleapps.io/api/1.1/obj",
    api_key="your-api-key",
    retry=retry_policy,
)

Usage in Sync Contexts

This library is async-only, but you can use it in sync code:

import asyncio
from bubble_data_api_client import BubbleModel, constraint, ConstraintType

class User(BubbleModel, typename="user"):
    name: str
    email: str
    early_access_enabled: bool = False

# simple scripts
user = asyncio.run(User.get("1234567890x1234567890"))

# or wrap multiple operations
async def main():
    constraints = [
        constraint("is_verified", ConstraintType.EQUALS, True),
        constraint("account_type", ConstraintType.EQUALS, "premium"),
    ]
    users = await User.find(constraints=constraints)
    for user in users:
        user.early_access_enabled = True
        await user.save()

asyncio.run(main())

Error Handling

from bubble_data_api_client import OnMultiple
from bubble_data_api_client.exceptions import (
    BubbleError,              # base exception
    BubbleHttpError,          # HTTP errors
    BubbleUnauthorizedError,  # 401/403 responses
    MultipleMatchesError,     # create_or_update found duplicates (with on_multiple=ERROR)
    PartialFailureError,      # some batch operations failed
    InvalidBubbleUIDError,    # invalid UID format
    ConfigurationError,       # missing configuration
)

# get() returns None if not found
user = await User.get("1661531100253x688916634279608300")
if user is None:
    print("User not found")

# create_or_update raises MultipleMatchesError with on_multiple=ERROR
try:
    user, created = await User.create_or_update(
        match={"external_id": "ext-123"},
        create_data={"name": "Test"},
        update_data={"name": "Test"},
        on_multiple=OnMultiple.ERROR,
    )
except MultipleMatchesError as e:
    print(f"Found {e.count} duplicates for {e.match}")

FAQ

How do I connect to a Bubble.io app from Python?

Install the package, then call configure() with your Bubble Data API root URL and API key. See Quick Start. The Data API must be enabled in your Bubble app under Settings → API.

How do I query Bubble.io records by field value from Python?

Use find() (or find_all() / find_iter()) with a list of constraint(...) objects. Each constraint takes a field name, a ConstraintType, and a value. See Constraints for the full operator list.

How do I handle Bubble Data API pagination?

The library handles pagination for you. Use find_all() to collect every matching record into a list, or find_iter() to stream records with constant memory. Both walk all pages internally. Use find() only if you want manual cursor / limit control. See Querying Records.

Does this support upserts?

Yes. create_or_update() matches by any field, creates if missing, updates if found, and offers configurable strategies for handling Bubble's lack of unique constraints (error, update first, update all, dedupe oldest, dedupe newest). See Smart Upserts.

Can I use this with FastAPI, Starlette, or other async frameworks?

Yes. The library is async-first and reuses HTTP connections per event loop, so it drops into any asyncio-based framework without extra configuration. Call the model methods directly from your route handlers.

Can I use this in synchronous Python code?

Yes, by wrapping calls in asyncio.run() or running an async block. See Usage in Sync Contexts.

How do I handle Bubble.io rate limits and retries?

Pass a tenacity.AsyncRetrying policy to configure(retry=...). You control the wait strategy, attempt count, and which exceptions to retry. See Retry Configuration.

What Python versions are supported?

Python 3.12 and newer. The library uses modern type-hint syntax and async features that require 3.12+.

License

MIT