boostspace-sdk 0.1.3

Official Boost.space API SDK (Python)

boostspace-sdk

Official Boost.space API SDK for Python — fully typed (pydantic v2 + httpx), generated from the OpenAPI spec. Sync and async.

Install

pip install boostspace-sdk

Requires Python ≥ 3.11.

Quickstart (sync)

from boostspace import BoostSpace

# Per-tenant base URL: https://<system>.boost.space/api
with BoostSpace(token="...", system="acme") as bs:
    space = bs.spaces.get(1)             # typed pydantic model

    for s in bs.spaces.list(limit=50):   # transparent pagination
        print(s.name)

    page = bs.spaces.list().page()       # single page + total
    print(page.found_records)

Quickstart (async)

import asyncio
from boostspace import AsyncBoostSpace

async def main() -> None:
    async with AsyncBoostSpace(token="...", system="acme") as bs:
        record = await bs.records.ref(1).get()
        labels = await bs.records.ref(1).labels().all()   # async pagination
        async for r in bs.records.list(limit=50):
            print(r.id)

asyncio.run(main())

OAuth2 / MCP tokens

Besides a static API token, the client accepts a callable token provider, resolved per request. OAuth2Session implements the Boost.space OAuth2 flow (MCP clients included — their tokens are ordinary OAuth2 access tokens) and is itself callable, refreshing proactively before expiry:

from boostspace.runtime import OAuth2Session

session = OAuth2Session("client-id", "client-secret", system="acme",
                        refresh_token=stored_refresh_token)
# or bootstrap from an authorization code:
# session.exchange_code(code, redirect_uri="https://app.example.com/cb")

bs = BoostSpace(session, system="acme")   # token stays fresh automatically

Relationship navigation

creator = bs.records.created_by(record)     # belongsTo → related record (User)
for label in bs.records.labels(record.id):   # hasMany → related collection
    ...

expanded = bs.records.expand(record, ["space", "created_by"])   # eager-load several at once
print(expanded.space.name if expanded.space else None)

Filtering

from boostspace import Filter

page = bs.spaces.list(
    filter=str(
        Filter.equals("color", "#4A90D9")
        .and_(Filter.greater_than_or_equal("id", 3))
        .and_(Filter.any_of(Filter.equals("name", "Acme"), Filter.equals("name", "Globex")))
    )
).page()
# color=#4A90D9;id>=3;(name=Acme|name=Globex)

Operators: eq ne gt gte lt lte like not_like in_ is_null, combined with and_/or_ (OR nested in AND is parenthesised automatically).

Upsert

Create-or-update by a match filter in a single call:

from boostspace import Filter, models

saved = bs.spaces.upsert(
    Filter.equals("name", "VIP"),
    models.Space(name="VIP", color="#ff0000"),
)

upsert lists by the filter, then updates the single match or creates one when none exists. More than one match raises AmbiguousMatchError. Available on resources that expose list + create + update over the same model.

Delta sync

Pull only the records changed since your last sync — pass updated_since (a Unix timestamp) to list, and pair it with list_deleted to learn which records were removed since the same point:

since = "1704067200"                                  # Unix timestamp of your last sync
for record in bs.records.list(updated_since=since):   # streams changed records
    reindex(record)
for deleted_id in bs.records.list_deleted(updated_since=since):
    drop(deleted_id)

Bulk with chunking

Split a large bulk update/delete into chunks and collect per-chunk failures instead of failing all-or-nothing:

report = bs.records.update_chunked(batch, chunk_size=100)
if not report.ok:
    for f in report.failed:  # BulkChunkError(index, size, error)
        print(f.index, f.error)
print(len(report.results))

Resolve a boostId

Map a Boost.space boostId (e.g. "Space5") to its typed record across entities, without knowing which resource it belongs to:

record = bs.resolve("Space5")   # a typed model, or None
if record is not None:
    print(record.boostId)

Returns None for an unknown prefix, a missing record, or a UUID-style id. The result is verified against the input boostId, so it is never a wrong-type record.

Field values

Read a record's Boost.space field values keyed by field name (self-contained from the record — no extra request):

values = bs.records.field_values(record)  # {"Rating": "5", "Tier": "gold"}

Available on resources whose model carries custom field values. Writing field values by name is planned for a later release.

Timeouts & observability

from boostspace import Hooks

bs = BoostSpace(
    token="...", system="acme",
    timeout=10.0,   # per-request timeout in seconds (default 30)
    hooks=Hooks(
        on_request=lambda m, u, h: h.__setitem__("X-Trace-Id", trace_id),
        on_response=lambda m, u, s, ms: metrics.record(s, ms),
    ),
)

Errors

from boostspace import NotFoundError

try:
    bs.spaces.get(999)
except NotFoundError as err:
    print(err.http_status, err.code)  # internal Boost.space error code

Requests retry with jittered backoff on 429/5xx, only for idempotent methods.

Testing without a live API

Inject an httpx client backed by a mock transport to unit-test your code:

import httpx
from boostspace import BoostSpace

def handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(200, json={"id": 1, "name": "Mock Inc"})

bs = BoostSpace("token", system="acme",
                http_client=httpx.Client(transport=httpx.MockTransport(handler)))
assert bs.spaces.get(1).name == "Mock Inc"  # no network

License

MIT

Guides

In-depth how-to guides live in docs/guides/: filtering · ordering & pagination · authentication · record handles (ref) · relations & expand · delta sync & bulk · errors & retries · testing & mocking