bubble-data-api-client 0.4.0

Python client for Bubble Data API

bubble-data-api-client

A fast, async Python client for the Bubble Data API with a Pydantic-based ORM and connection pooling.

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
users = await User.find(constraints=[
    constraint("status", ConstraintType.EQUALS, "active")
])

# 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 and delete the rest
user, created = await User.create_or_update(
    match={"external_id": "ext-123"},
    data={"name": "Canonical Name"},
    on_multiple=OnMultiple.DEDUPE_OLDEST,
)

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
• 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.13+.

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
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),
])

# 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 - match by external_id, create if not found
user, created = await User.create_or_update(
    match={"external_id": "ext-123"},
    data={"name": "Updated Name", "email": "[e-mail]"},
    on_multiple=OnMultiple.ERROR,
)
# returns (User, bool) - the model instance and whether it was created

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	Keep oldest record, delete others, then update
OnMultiple.DEDUPE_NEWEST	Keep newest record, delete others, then update

# auto-deduplicate, keeping the oldest record
user, created = await User.create_or_update(
    match={"external_id": "ext-123"},
    data={"name": "Canonical Name"},
    on_multiple=OnMultiple.DEDUPE_OLDEST,
)

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.

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"},
        data={"name": "Test"},
        on_multiple=OnMultiple.ERROR,
    )
except MultipleMatchesError as e:
    print(f"Found {e.count} duplicates for {e.match}")

License

MIT