Supply Chain Management with Uni¶

This notebook demonstrates how to model a supply chain graph to perform BOM (Bill of Materials) explosion, cost rollup, and supplier risk analysis.

import os
import shutil
import tempfile

import uni_db

db_path = os.path.join(tempfile.gettempdir(), "supply_chain_db")
if os.path.exists(db_path):
    shutil.rmtree(db_path)
db = uni_db.Uni.open(db_path)
session = db.session()
print(f"Opened database at {db_path}")

Opened database at /tmp/supply_chain_db

1. Define Schema¶

Parts, Suppliers, and Products with assembly and supply relationships.

(
    db.schema()
    .label("Part")
    .property("name", "string")
    .property("sku", "string")
    .property("cost", "float64")
    .index("sku", "hash")
    .done()
    .label("Supplier")
    .property("name", "string")
    .done()
    .label("Product")
    .property("name", "string")
    .property("price", "float64")
    .done()
    .edge_type("ASSEMBLED_FROM", ["Product", "Part"], ["Part"])
    .done()
    .edge_type("SUPPLIED_BY", ["Part"], ["Supplier"])
    .done()
    .apply()
)

print("Schema created")

Schema created

2. Ingest Data¶

Two products sharing common parts, supplied by multiple vendors.

tx = session.tx()
with tx.bulk_writer().build() as bw:
    # Parts: 7 components with different costs
    part_vids = bw.insert_vertices(
        "Part",
        [
            {"name": "Resistor 10K", "sku": "RES-10K", "cost": 0.05},
            {"name": "Capacitor 100uF", "sku": "CAP-100UF", "cost": 0.08},
            {"name": "Motherboard X1", "sku": "MB-X1", "cost": 50.0},
            {"name": "OLED Screen", "sku": "SCR-OLED", "cost": 30.0},
            {"name": "Battery 4000mAh", "sku": "BAT-4000", "cost": 15.0},
            {"name": "ARM Processor", "sku": "PROC-ARM", "cost": 80.0},
            {"name": "LCD Screen", "sku": "SCR-LCD", "cost": 20.0},
        ],
    )
    res10k, cap100uf, mbx1, scr_oled, bat4000, proc_arm, scr_lcd = part_vids

    # Suppliers
    sup_vids = bw.insert_vertices(
        "Supplier",
        [
            {"name": "ResistorWorld"},
            {"name": "ScreenTech"},
            {"name": "CoreComponents"},
        ],
    )
    resistor_world, screen_tech, core_components = sup_vids

    # Products
    prod_vids = bw.insert_vertices(
        "Product",
        [
            {"name": "Smartphone X", "price": 599.0},
            {"name": "TabletPro 10", "price": 799.0},
        ],
    )
    smartphone, tablet = prod_vids

    # Assembly: Smartphone X -> OLED screen, shared parts
    bw.insert_edges(
        "ASSEMBLED_FROM",
        [
            (smartphone, mbx1, {}),
            (smartphone, scr_oled, {}),
            (smartphone, bat4000, {}),
            (smartphone, proc_arm, {}),
            (mbx1, res10k, {}),
            (mbx1, cap100uf, {}),
        ],
    )

    # Assembly: TabletPro 10 -> LCD screen, shared parts
    bw.insert_edges(
        "ASSEMBLED_FROM",
        [
            (tablet, mbx1, {}),
            (tablet, scr_lcd, {}),
            (tablet, bat4000, {}),
            (tablet, proc_arm, {}),
        ],
    )

    # Supply relationships
    bw.insert_edges(
        "SUPPLIED_BY",
        [
            (res10k, resistor_world, {}),
            (cap100uf, resistor_world, {}),
            (scr_oled, screen_tech, {}),
            (scr_lcd, screen_tech, {}),
            (mbx1, core_components, {}),
            (bat4000, core_components, {}),
            (proc_arm, core_components, {}),
        ],
    )

    bw.commit()
tx.commit()
print("Data ingested")

Data ingested

3. BOM Explosion¶

Which products are affected if RES-10K is defective? Traverses the assembly hierarchy upward.

query_bom = """
    MATCH (defective:Part {sku: 'RES-10K'})
    MATCH (product:Product)-[:ASSEMBLED_FROM*]->(defective)
    RETURN product.name AS name, product.price AS price
    ORDER BY product.price DESC
"""
results = session.query(query_bom)
print("Products affected by defective RES-10K:")
for r in results:
    print(f"  {r['name']} (${r['price']})")
assert len(results) == 2, f"Expected 2 affected products, got {len(results)}"

Products affected by defective RES-10K:
  TabletPro 10 ($799.0)
  Smartphone X ($599.0)

Bounded vs Unbounded Paths: [*] performs unbounded traversal (defaults to 100 hops max), ideal for BOM explosion where you want every affected product regardless of depth. Use [*1..5] to cap traversal at a known depth, as shown in the queries below.

4. Full BOM Listing¶

Every part in Smartphone X with its cost, ordered by cost descending.

query_parts = """
    MATCH (p:Product {name: 'Smartphone X'})-[:ASSEMBLED_FROM*1..5]->(part:Part)
    RETURN part.name AS part_name, part.sku AS sku, part.cost AS cost
    ORDER BY cost DESC
"""
results = session.query(query_parts)
print("Smartphone X BOM:")
for r in results:
    print(f"  {r['part_name']} ({r['sku']}): ${r['cost']}")

Smartphone X BOM:
  ARM Processor (PROC-ARM): $80.0
  Motherboard X1 (MB-X1): $50.0
  OLED Screen (SCR-OLED): $30.0
  Battery 4000mAh (BAT-4000): $15.0
  Capacitor 100uF (CAP-100UF): $0.08
  Resistor 10K (RES-10K): $0.05

5. Cost Rollup¶

Total BOM cost per product — GROUP BY product with SUM of part costs.

query_rollup = """
    MATCH (p:Product)-[:ASSEMBLED_FROM*1..5]->(part:Part)
    RETURN p.name AS product, SUM(part.cost) AS total_bom_cost
    ORDER BY total_bom_cost DESC
"""
results = session.query(query_rollup)
print("BOM cost rollup per product:")
for r in results:
    print(f"  {r['product']}: ${r['total_bom_cost']:.2f}")
assert len(results) == 2, f"Expected 2 rows, got {len(results)}"

BOM cost rollup per product:
  Smartphone X: $175.13
  TabletPro 10: $165.13

6. Supply Chain Risk¶

Which supplier is critical to the most products?

query_risk = """
    MATCH (p:Product)-[:ASSEMBLED_FROM*1..5]->(part:Part)-[:SUPPLIED_BY]->(s:Supplier)
    RETURN s.name AS supplier, COUNT(DISTINCT p) AS products_at_risk
    ORDER BY products_at_risk DESC
"""
results = session.query(query_risk)
print("Supplier risk analysis:")
for r in results:
    print(f"  {r['supplier']}: {r['products_at_risk']} product(s) at risk")
assert len(results) == 3, f"Expected 3 suppliers, got {len(results)}"
top = results[0]
assert top["products_at_risk"] == 2, f"Expected 2, got {top['products_at_risk']}"

Supplier risk analysis:
  ScreenTech: 2 product(s) at risk
  CoreComponents: 2 product(s) at risk
  ResistorWorld: 2 product(s) at risk