Metadata-Version: 2.4
Name: bubble-data-api-client
Version: 0.7.2
Summary: 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.
License-Expression: MIT
Project-URL: Homepage, https://github.com/bubble-python/bubble-data-api-client
Project-URL: Repository, https://github.com/bubble-python/bubble-data-api-client
Keywords: bubble,bubble.io,bubble-api,bubbleio,async,asyncio,pydantic,orm,api,sdk,client,data-api,rest-api,typed,crud,database,integration,etl,httpx,low-code,nocode
Classifier: Framework :: AsyncIO
Classifier: Framework :: Pydantic
Classifier: Framework :: Pydantic :: 2
Classifier: Intended Audience :: Developers
Classifier: Natural Language :: English
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx[http2]~=0.28
Requires-Dist: pydantic~=2.12
Requires-Dist: tenacity~=9.1
Requires-Dist: typing_extensions>=4.10
Dynamic: license-file


# bubble-data-api-client

[![PyPI](https://img.shields.io/pypi/v/bubble-data-api-client)](https://pypi.org/project/bubble-data-api-client/)
[![Python Version](https://img.shields.io/pypi/pyversions/bubble-data-api-client)](https://pypi.org/project/bubble-data-api-client/)
[![License](https://img.shields.io/pypi/l/bubble-data-api-client)](https://pypi.org/project/bubble-data-api-client/)
[![CI](https://img.shields.io/github/actions/workflow/status/bubble-python/bubble-data-api-client/test.yml?branch=main&label=tests)](https://github.com/bubble-python/bubble-data-api-client/actions/workflows/test.yml)
[![Downloads](https://static.pepy.tech/badge/bubble-data-api-client/month)](https://pepy.tech/project/bubble-data-api-client)
[![Pydantic v2](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/pydantic/pydantic/main/docs/badge/v2.json)](https://pydantic.dev)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)

**Query your Bubble.io database from Python.** A fast, async client and Pydantic ORM for the [Bubble Data API](https://manual.bubble.io/core-resources/api/the-bubble-api/the-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

```python
# create
user = await User.create(name="Ada", email="ada@example.com")

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

```python
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:

```python
# Type mismatch caught immediately
user = 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 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:

```python
# 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

```bash
pip install bubble-data-api-client
```

Requires Python 3.12+.

## Quick Start

### Configuration

```python
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:

```python
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:

```python
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:

```python
# create
user = await User.create(name="Ada Lovelace", email="ada@example.com")

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

```python
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": "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 |

```python
# 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:

```python
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 |

```python
# 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:

```python
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:

```python
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="test@example.com")

# option 2: manual cleanup
await close_clients()
```

## Retry Configuration

Plug in custom retry policies using `tenacity`:

```python
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:

```python
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

```python
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](#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](#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](#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](#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](#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](#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
