How do I push historic data to Entities in Python?


Historic records let an Entity keep its previous values against a date and time, so you can ask what it looked like at any point rather than only what it looks like now. Backfilling readings from another system is the usual reason to write them yourself.

First, connect an attribute to a date-time

History is keyed on one of your own date-time attributes. In the Entity Type's Data Schema, set HistoricKey on each attribute you want tracked, pointing at that date-time attribute. That date-time is the historic key, and the attributes pointing at it are the historic attributes.

Nothing is recorded until that link exists, so this is the step to check first when a write appears to do nothing. See the Entity historic data documentation for the full definition.

{
    "Key": "root",
    "Name": "Root",
    "Type": "Structure",
    "Structure": [
        {
            "Key": "reading_taken",
            "Name": "Reading taken",
            "Type": "Datetime"
        },
        {
            "Key": "temperature",
            "Name": "Temperature",
            "Type": "Double",
            "HistoricKey": "reading_taken"
        }
    ]
}

For an attribute nested inside a Structure, the HistoricKey is slash separated from the root. Eg: readings/taken.

With that in place, updating an Entity with a new value in the date-time attribute records the previous state automatically. The rest of this page is about writing records directly, which is what a backfill needs.


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

pip install bruce-models

Python
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 date-time attribute history is keyed on, as path segments.
HISTORIC_KEY = ["reading_taken"]

BATCH_SIZE = 500

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

# Readings to load, keyed by the Entity they belong to.
# In a real backfill this comes from your own system.
readings_by_entity = {
    "entity-id-one": [
        ("2026-01-01T09:00:00Z", 21.4),
        ("2026-01-02T09:00:00Z", 22.1)
    ],
    "entity-id-two": [
        ("2026-01-01T09:00:00Z", 18.9)
    ]
}

def build_records(readings_by_entity):
    """
    Turns the readings into historic records.
    """
    records = []
    for entity_id, readings in readings_by_entity.items():
        for date_time, temperature in readings:
            records.append(BM.EntityHistoricData.build_record(
                entity_id,
                HISTORIC_KEY,
                date_time,
                # The recorded values are shaped like the Entity itself.
                {
                    "reading_taken": date_time,
                    "temperature": temperature
                }
            ))
    return records

def push(bruce, records):
    """
    Writes the records a batch at a time.
    """
    written = 0
    for start in range(0, len(records), BATCH_SIZE):
        batch = records[start:start + BATCH_SIZE]
        res = BM.EntityHistoricData.update(bruce, batch)
        written += len(res.get("Items") or [])

        # A record that could not be written is reported rather than failing the request.
        for warning in (res.get("Warning") or []):
            print(warning)

    return written

records = build_records(readings_by_entity)
print(f"pushing {len(records)} records")
print(f"wrote {push(bruce, records)}")

# Reading them back, which returns the same shape that went in.
for record in BM.EntityHistoricData.iterate(
    bruce,
    list(readings_by_entity.keys()),
    attr_key=HISTORIC_KEY
):
    values = record.get("Entity") or {}
    print(record.get("Entity.ID"), record.get("DateTime"), values.get("temperature"))

Recording where the Entity was, too

A record keeps a full snapshot of the platform managed values as well as your attributes, so a moving Entity can keep its past positions. Pass them as internal and they are nested under the record's Bruce key.

record = BM.EntityHistoricData.build_record(
    "entity-id-one",
    ["surveyed"],
    "2026-01-01T09:00:00Z",
    {"surveyed": "2026-01-01T09:00:00Z"},
    internal={"Location": {"latitude": -36.8485, "longitude": 174.7633, "altitude": 0}}
)

Reading it back as the Entity

Once written, the records are what a point-in-time read overlays. Ask for an Entity at a date and time and the nearest matching record replaces its values, reported in the Outline like any other source.

entity = BM.Entity.get(
    bruce,
    "entity-id-one",
    historic_point="2026-01-01T12:00:00Z"
)

# The Outline says which record was used and what it contributed.
for source in (entity.get("Bruce").get("Outline") or []):
    if source.get("DateTime"):
        print("values from", source.get("DateTime"), "keyed on", source.get("ForeignKey"))

Things worth knowing

Write the key attribute into the values. The date-time named by the historic key should appear in the recorded values as well as in the record's own DateTime. A record missing it still saves, but reading the Entity back at that point in time will not show the date it came from.

There is a record limit per Entity. The default is 1,000 records per Entity ID unless an upgraded limit is discussed. Past that, the oldest are dropped, so a long backfill at a fine interval can quietly evict its own earliest records.

Writing a record does not change the Entity. Its current values are untouched, which is the point. Update the Entity itself if you also want the latest state to move.

Scenarios. Passing a scenario writes the series as a variant rather than as real world data, and a Scenario identifies its own records, so the date-time and attribute become optional there.

Removing a bad backfill. Two options. delete_records takes the records themselves, so the ones you just wrote or just read are the ones removed. delete takes the same filter as a read, so the range you wrote is the range you clear; it requires the attribute and both ends of the range, which is what stops an unbounded delete.

# The write response returns what landed, so it can be handed straight back.
res = BM.EntityHistoricData.update(bruce, records)
BM.EntityHistoricData.delete_records(bruce, res.get("Items") or [])