A Delta Catalog Table

"""BookNest's XML catalog (Chapter 2) as a Delta table on MinIO, then a price feed merged."""
import subprocess, xml.etree.ElementTree as ET
import pyarrow as pa
from deltalake import DeltaTable, write_deltalake
from rs_conf import S3
URI = "s3://warehouse/delta/catalog"
xml = subprocess.run(["mc", "cat", "lake/booknest-raw/catalog/booknest-catalog.xml"],
                     capture_output=True, check=True).stdout
books = pa.Table.from_pylist([{"sku": b.findtext("identifier"), "title": b.findtext("title"),
    "genre": b.findtext("subject"), "price": b.findtext("supply/price")}
    for b in ET.fromstring(xml).iter("book")])
books = books.set_column(3, "price", books["price"].cast(pa.decimal128(9, 2)))  # not float
write_deltalake(URI, books, mode="overwrite", storage_options=S3)
feed = pa.table({"sku": ["BN-0002", "BN-0005"],                       # tomorrow's price feed
                 "price": pa.array(["34.50", "16.20"]).cast(pa.decimal128(9, 2))})
m = (DeltaTable(URI, storage_options=S3)
     .merge(feed, "t.sku = s.sku", source_alias="s", target_alias="t")
     .when_matched_update({"price": "s.price"}, predicate="t.price <> s.price")
     .execute())
print({k: v for k, v in m.items() if k.startswith("num_target_rows")})

{'num_target_rows_inserted': 0, 'num_target_rows_updated': 1, 'num_target_rows_deleted': 0,
  'num_target_rows_copied': 5}