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

# createuser=awaitUser.create(name="Ada", email="ada@example.com")
# retrieveuser=awaitUser.get(uid)
# query (paginated)users=awaitUser.find(
constraints=[
constraint("status", ConstraintType.EQUALS, "active"),
]
)
# query all matching recordsall_users=awaitUser.find_all()
# iterate with constant memoryasyncforuserinUser.find_iter():
process(user)
# updateuser.name="Ada Lovelace"awaituser.save()
# deleteawaituser.delete()
# check existenceifawaitUser.exists(uid):
print("User exists")
# countactive_count=awaitUser.count(
constraints=[
constraint("status", ConstraintType.EQUALS, "active"),
]
)

IDE support and type checking

Models provide autocomplete and catch errors before runtime:

classUser(BubbleModel, typename="user"):
name: stremail: strage: intuser=awaitUser.get(uid)
user.name# IDE autocomplete worksuser.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 immediatelyuser=User(
_id="123x456",
name="Ada",
email="ada@example.com",
age="twenty-five",
)
# ValidationError: Input should be a valid integer# Invalid Bubble UID caught at the model levelclassOrder(BubbleModel, typename="order"):
customer: BubbleUIDorder=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. Every model also inherits Bubble's other built-in fields automatically: created_by, created_date, modified_date, and slug. All are populated from API responses, and save() never writes any of them back: the dates are server-managed, created_by is set at creation, and the Data API rejects slug on update
• 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 restuser, created=awaitUser.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
• Unlimited scanning:scan() streams collections of any size, past Bubble's ~50,000 record limit
• 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

frombubble_data_api_clientimportconfigureconfigure(
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:

importosfrombubble_data_api_clientimportset_config_provider, BubbleConfigdefget_config() ->BubbleConfig:
returnBubbleConfig(
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:

frombubble_data_api_clientimportBubbleModel, BubbleUID, OptionalBubbleUIDclassUser(BubbleModel, typename="user"):
name: stremail: strcompany: OptionalBubbleUID=None# linked Bubble recordclassCompany(BubbleModel, typename="company"):
name: strindustry: str

Then use them:

# createuser=awaitUser.create(name="Ada Lovelace", email="ada@example.com")
# retrieveuser=awaitUser.get("1234567890x1234567890")
# query with constraints (single page)frombubble_data_api_clientimportconstraint, ConstraintTypeactive_users=awaitUser.find(
constraints=[
constraint("status", ConstraintType.EQUALS, "active"),
constraint("age", ConstraintType.GREATER_THAN, 18),
]
)
# get all matching records as a listall_active=awaitUser.find_all(
constraints=[
constraint("status", ConstraintType.EQUALS, "active"),
]
)
# iterate through all records with constant memoryasyncforuserinUser.find_iter():
print(user.name)
# updateuser.name="Ada L."awaituser.save()
# deleteawaituser.delete()

Smart Upserts

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

frombubble_data_api_clientimportOnMultiple# basic upsert, matches by external_id and creates if not founduser, created=awaitUser.create_or_update(
match={"external_id": "ext-123"},
create_data={"name": "New User", "email": "new@example.com"},
update_data={"email": "new@example.com"},
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 Dateuser, created=awaitUser.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:

frombubble_data_api_clientimportconstraint, ConstraintTypeconstraints= [
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=awaitUser.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.

Field Name Introspection

When a model declares Field(alias=...), bubble_field() returns the Bubble field name for a Python attribute, avoiding restating alias strings at every call site:

frompydanticimportFieldclassUser(BubbleModel, typename="user"):
first_name: str|None=Field(default=None, alias="firstName")
last_name: str|None=Field(default=None, alias="lastName")
User.bubble_field("first_name") # "firstName"User.bubble_field("created_date") # "Created Date"User.bubble_field("typo") # raises UnknownFieldError

For fields without an alias, the Python attribute name is returned unchanged.

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
scan()	AsyncIterator	Stream collections of any size, past Bubble's ~50,000 record limit

# find(): single page, you control paginationpage1=awaitUser.find(limit=100)
page2=awaitUser.find(limit=100, cursor=100)
# find_all(): fetches all pages, returns when completeall_users=awaitUser.find_all(constraints=[...])
print(f"Got {len(all_users)} users")
# find_iter(): streams records with constant memoryasyncforuserinUser.find_iter(constraints=[...]):
awaitprocess(user) # each record processed as it arrives# scan(): streams records of any size, past the ~50k offset capasyncforuserinUser.scan(constraints=[...]):
awaitprocess(user)

Both find_all() and find_iter() handle pagination internally, fetching pages of page_size (default 100) until all records are retrieved. For very large collections, past Bubble's ~50,000 record pagination limit, use scan() (see Scanning Large Collections).

Scanning Large Collections

Bubble stops paginating once you reach roughly 50,000 records, so find_all() and find_iter() can't walk a bigger collection in full. scan() removes that ceiling: it streams every record, no matter how large the collection, with constant memory.

# stream every record, however many there areasyncforuserinUser.scan():
awaitprocess(user)
# filter with the same constraints as find()asyncforuserinUser.scan(
constraints=[
constraint("status", ConstraintType.EQUALS, "active"),
]
):
awaitprocess(user)
# fetch pages in parallel for higher throughput (default is 1, sequential)asyncforuserinUser.scan(concurrency=10):
awaitprocess(user)
# also available on the raw client, yielding plain dictsasyncwithRawClient() asclient:
asyncforrowinclient.scan("user"):
process(row)

Records arrive in Created Date order, the one trade-off for unlimited iteration. Pass keyset_field=... to order by a different date field. For collections that fit under the cap, find_iter() stays the simpler choice and supports any sort order.

Sequential scanning is bound by Bubble's per-page latency (often around one second per 100 records). concurrency=N fetches up to N pages in parallel while preserving ordering and the no-duplicate guarantee; throughput scales near-linearly with N. It also multiplies your request rate against Bubble, so pick a value your plan's API limits allow.

Type-Safe Bubble UIDs

Validate Bubble record IDs at the type level:

frombubble_data_api_clientimport (
BubbleModel,
BubbleUID,
OptionalBubbleUID,
OptionalBubbleUIDs,
)
classOrder(BubbleModel, typename="order"):
customer: BubbleUID# required, validatedreferrer: OptionalBubbleUID=None# optional, coerces invalid to Noneitems: OptionalBubbleUIDs=None# list of UIDs, filters invalid# validation helpersfrombubble_data_api_clientimportis_bubble_uid, filter_bubble_uidsis_bubble_uid("1234567890x1234567890") # Trueis_bubble_uid("invalid") # Falseuids= ["1661531100253x688916634279608300", "invalid", None]
filter_bubble_uids(uids) # ["1661531100253x688916634279608300"]

Connection Pooling

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

frombubble_data_api_clientimportclient_scope, close_clients# option 1: context manager (auto-closes on exit)asyncwithclient_scope():
awaitUser.create(name="Test", email="test@example.com")
# option 2: manual cleanupawaitclose_clients()

Retry Configuration

Plug in custom retry policies using tenacity:

importhttpximporttenacityfrombubble_data_api_clientimportconfigureretry_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:

importasynciofrombubble_data_api_clientimportBubbleModel, constraint, ConstraintTypeclassUser(BubbleModel, typename="user"):
name: stremail: strearly_access_enabled: bool=False# simple scriptsuser=asyncio.run(User.get("1234567890x1234567890"))
# or wrap multiple operationsasyncdefmain():
constraints= [
constraint("is_verified", ConstraintType.EQUALS, True),
constraint("account_type", ConstraintType.EQUALS, "premium"),
]
users=awaitUser.find(constraints=constraints)
foruserinusers:
user.early_access_enabled=Trueawaituser.save()
asyncio.run(main())

Error Handling

frombubble_data_api_clientimportOnMultiplefrombubble_data_api_client.exceptionsimport (
BubbleError, # base exceptionBubbleHttpError, # HTTP errorsBubbleUnauthorizedError, # 401/403 responsesMultipleMatchesError, # create_or_update found duplicates (with on_multiple=ERROR)PartialFailureError, # some batch operations failedInvalidBubbleUIDError, # invalid UID formatConfigurationError, # missing configuration
)
# get() returns None if not founduser=awaitUser.get("1661531100253x688916634279608300")
ifuserisNone:
print("User not found")
# create_or_update raises MultipleMatchesError with on_multiple=ERRORtry:
user, created=awaitUser.create_or_update(
match={"external_id": "ext-123"},
create_data={"name": "Test"},
update_data={"name": "Test"},
on_multiple=OnMultiple.ERROR,
)
exceptMultipleMatchesErrorase:
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.

How do I paginate past Bubble's 50,000 record limit?

Use scan(). It streams every record in a collection of any size, where find_all() and find_iter() stop at Bubble's ~50,000 record limit. Records come back in Created Date order. See Scanning Large Collections.

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