Loading BookNest Data with dlt

The order export again, now with dlt 876,044 (ingest/dlt/, in its own Python 3.14 virtual environment): June's orders merged on order_id, with order_ts as an incremental cursor so each run loads only newer orders:

@dlt.resource(primary_key="order_id", write_disposition="merge")
def orders(order_ts=dlt.sources.incremental("order_ts", initial_value="2026-06-01T00:00:00Z")):
    """Orders from June 2026 on; dlt saves the newest order_ts as state for the next run."""
    with open(EXPORT, encoding="utf-8") as f:
        for line in f:
            yield json.loads(line)               # dlt drops rows older than the cursor
pipeline = dlt.pipeline(
    pipeline_name="booknest_orders",
    destination=dlt.destinations.postgres(
        "postgresql://postgres:booknest@localhost:32543/booknest"),
    dataset_name="dlt_orders",                   # the PostgreSQL schema dlt creates
)
info = pipeline.run(orders())
print(pipeline.last_trace.last_normalize_info.row_counts)

run.sh ran it three times; before the third run it appended one late sample order for 1 July to the work copy of the export (and restored the file afterwards):

--- run 1
{'orders': 5498, 'orders__items': 7546, '_dlt_pipeline_state': 1}
cursor now: 2026-06-30T23:55:02Z
--- run 2
{}
cursor now: 2026-06-30T23:55:02Z
--- run 3
{'_dlt_pipeline_state': 1, 'orders': 1, 'orders__items': 1}
cursor now: 2026-07-01T00:04:10Z

dlt created dlt_orders.orders and the child table orders__items (book_id, qty, unit_price, plus _dlt_parent_id and _dlt_list_idx), typed order_ts as timestamp with time zone, and held 5,499 orders after the third run. The second run read the whole file (6-8 seconds on this shared host) and loaded nothing; the cursor only filters. Two things to fix before production: money arrived as double precision, because JSON numbers are floats to dlt, so declare columns={"total": {"data_type": "decimal"}} and the like; and a cursor on order_ts misses orders whose status changes later, which is what CDC (CDC with Debezium 3.x) or a merge on an updated_at cursor is for.