Use Data Hub from Python

Read and write Data Hub records from a Python automation with velophex.runtime.

Beta

Data Hub is in Beta. It is off by default; an administrator turns it on per tenant.

Python automations use velophex.runtime. Data Hub calls are in the velophex SDK from version 0.3.0. An entity is addressed by its API name and its fields by theirs. See Python SDK.

A first automation#

You need the Invoice entity from the overview and a personal access token.

PowerShellCopy

uvx --python 3.12 velophex init Ledger
cd Ledger
uv sync

Replace src/ledger/main.py with:

src/ledger/main.pyPythonCopy

"""Invoice ledger: put an invoice in Data Hub, then read the open ones back."""

from decimal import Decimal

import velophex
from velophex import runtime


@velophex.entrypoint(id="main", display_name="Run")
def run(invoice_number: str = "INV-1001", amount: str = "1250.00") -> dict[str, object]:
    """Create the invoice, or update the one with the same number, and list the open invoices."""
    result = runtime.upsert_record(
        "Invoice",
        "invoiceNumber",
        {"invoiceNumber": invoice_number, "amount": Decimal(amount), "status": "open"},
    )

    open_invoices = [
        f"{invoice['invoiceNumber']} {invoice['amount']:.2f}"
        for invoice in runtime.iter_records("Invoice", match={"status": "open"}, sort="amount desc")
    ]

    return {"created": result.created, "revision": result.record.revision, "open": open_invoices}

Run it against the Orchestrator. This is a design-time run, the Python twin of F5 in Studio:

PowerShellCopy

$env:VELOPHEX_API_KEY = "<personal access token>"
uv run velophex run main --orchestrator https://orchestrator.example.com --tenant <tenant id> --workspace Finance

The first run prints:

JSONCopy

{
  "created": true,
  "revision": 1,
  "open": [
    "INV-1001 1250.00"
  ]
}

A second run prints "created": false and "revision": 2.

Calls#

Call	What it does
`list_entities()`, `entity(name)`, `entity_json_schema(name)`	The workspace's entities, one entity's fields, the JSON Schema of a record
`query_records(entity, filter, *, match, sort, select, page_size, cursor, include_total)`	One page of records; `next_cursor` for the next. `page_size` up to 1,000.
`iter_records(entity, filter, *, ..., limit=10_000)`	Every match, page after page. Raises `DataHubResultTooLargeError` past `limit` or 64 MiB; `limit=None` lifts it.
`get_record(entity, id)`	One record
`create_record(entity, data, *, idempotency_key=None)`	A new record. The same key and data within 24 hours return the first one.
`upsert_record(entity, key_field, data)`	Creates, or updates the record with that unique key. `.created` says which.
`update_record(entity, id, values, *, revision=None)`	Changes only the fields given. `None` clears one. `revision` refuses a stale write.
`delete_record(entity, id)`	Deletes a record, if the entity allows automations to delete records

PythonCopy

from decimal import Decimal

from velophex import runtime
from velophex.runtime import where

result = runtime.upsert_record("Invoice", "invoiceNumber", {"invoiceNumber": "INV-23", "amount": Decimal("1204.50")})

for invoice in runtime.iter_records("Invoice", where("amount").gte(Decimal("1000")), sort="amount desc"):
    print(invoice["invoiceNumber"], invoice["amount"])  # amount is a Decimal

runtime.update_record("Invoice", result.record.id, {"status": "paid"}, revision=result.record.revision)

Filters, sort and values#

Filters are where("field").op(value), combined with & and |. The operators are eq, ne, in_, not_in, gt, gte, lt, lte, contains, starts_with, is_null and is_not_null. You can also pass a dict, or the JSON from the web app's Copy filter as JSON. match={"status": "open"} adds equalities.
Sort is one string, such as "amount desc, invoiceNumber". Only indexed and system fields sort.
Values are typed from the entity's schema: decimal is Decimal (sent as a string, never a float), date is date, date and time is an aware UTC datetime (a naive one is refused), number is int.
A record's system fields are attributes (id, revision, created_at and the others). Its values are record["name"].
A filter on a field that is not indexed raises a DataHubWarning on small entities. Above 50,000 records it is refused with FieldNotIndexedError.

Errors#

Refusals raise a DataHubError subclass with cause, fix and docs_url:

EntityNotFoundError, RecordNotFoundError
RecordValidationError (.errors by data.<field>), UniqueViolationError (.field)
RevisionConflictError, SchemaChangedError
FilterInvalidError (.pointer), CursorInvalidError, FieldNotIndexedError, QueryTooExpensiveError
KeyFieldNotUniqueError, IdempotencyKeyReusedError, DataHubLimitReachedError
DeleteDisabledError, DataHubAccessDeniedError
DataHubDisabledError when Data Hub is off for the tenant, and DataHubUnsupportedError when the Orchestrator does not offer it

When a cursor stops working you get CursorInvalidError. Start again from the first page, and keep the last id you processed if you need to resume.

A first automation

Calls

Filters, sort and values

Errors