How do I load and update Entities in batches in Python?

Reading an Entity Type in one request and writing it back in one request both stop working once the Type is large. The fix is the same in each direction: work a batch at a time.

The example below walks an Entity Type a page at a time, queues a change per record, and writes the queue out in batches at the end. A Change Set built locally holds the edits, merges repeated edits to one Entity, and spills to disk if the walk queues more than memory should hold, so the run stays flat no matter how large the Type is. A failure costs you one batch rather than the whole run, and the report names what did not land.

You will want to install our Python library to facilitate this process.

pip install bruce-models

import bruce_models as BM

ENTITY_TYPE_ID = "YOUR_ENTITY_TYPE_ID"
# Either a session token from a login or a long-lived access token.
TOKEN = "YOUR_TOKEN"

# The attribute being changed, as path segments.
ATTR_PATH = ["status"]

BATCH_SIZE = 500

bruce = BM.BruceApi({
    "account_id": "YOUR_ACCOUNT_ID",
    "session_id": TOKEN
})

# The selection to walk. Nothing is requested until you walk it.
query = BM.DataLabBuilder(bruce, page_size=BATCH_SIZE).entity_type(ENTITY_TYPE_ID)

# Collects the edits locally. Nothing is sent while it fills.
changes = BM.ChangeSetBuilder(bruce)

def consider(entity):
    """
    Queues a change for one Entity, or leaves it alone.
    """
    if BM.Entity.get_value(entity, ATTR_PATH) == "Active":
        return

    values = {}
    BM.Entity.set_value(values, ATTR_PATH, "Active")
    # Reads the ID and Entity Type off the record for you.
    changes.queue(entity, values)

def read(page):
    """
    Handles one page of Entities.
    """
    for entity in page.items:
        consider(entity)
    print(f"read page {page.number}, {page.seen:,} seen, "
          f"{len(changes):,} queued")

query.each_page(read)

def saving(state):
    """
    Reports each batch of writes as it settles.
    """
    print(f"[{state['percent']:>3}%] saved {state['processed']:,} "
          f"of {state['requested']:,}")

# Writes the queue out in batches. The API overlays what it is sent
# onto each record, so only the attribute above changes.
report = changes.apply(batch_size=BATCH_SIZE, on_progress=saving)

print(f"{report['succeeded']:,} updated, {report['failed_count']:,} failed")
for entity_id, reason in list(report["failed"].items())[:5]:
    print(f"  {entity_id}: {reason}")

Things worth knowing

Only send what changed. Writing back every record you read costs the same as writing back the handful that differ, and it re-indexes records that did not need it. Queueing nothing for a record that already holds the value, as above, is cheaper than deciding it server side.

Reading before writing. The example queues every change during the walk and writes afterwards, so nothing shifts underneath the paging. Writing as you go is faster to first result, but if the change stops a record matching the filter, the pages move while you are reading them.

Use set_value rather than assigning into the dictionary. Attributes address by quoted path, and some of them are mid-migration between two locations in the record. The helper writes the right one.

High volume writes. If you are pushing many batches back to back and do not need each response immediately, the API accepts a Defer flag that queues the save instead of blocking on it. Pass it as a request param when throughput matters more than confirmation.