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.

In [1]:

import os
import shutil
import tempfile

import uni_db

import os import shutil import tempfile import uni_db

In [2]:

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

db_path = os.path.join(tempfile.gettempdir(), "supply_chain_db") if os.path.exists(db_path): shutil.rmtree(db_path) db = uni_db.Database(db_path) 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.

In [3]:

(
    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")

( 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.

In [4]:

# Parts: 7 components with different costs
part_vids = db.bulk_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 = db.bulk_insert_vertices('Supplier', [
    {'name': 'ResistorWorld'},
    {'name': 'ScreenTech'},
    {'name': 'CoreComponents'},
])
resistor_world, screen_tech, core_components = sup_vids

# Products
prod_vids = db.bulk_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
db.bulk_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
db.bulk_insert_edges('ASSEMBLED_FROM', [
    (tablet, mbx1,    {}),
    (tablet, scr_lcd, {}),
    (tablet, bat4000, {}),
    (tablet, proc_arm, {}),
])

# Supply relationships
db.bulk_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, {}),
])

db.flush()
print("Data ingested")

# Parts: 7 components with different costs part_vids = db.bulk_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 = db.bulk_insert_vertices('Supplier', [ {'name': 'ResistorWorld'}, {'name': 'ScreenTech'}, {'name': 'CoreComponents'}, ]) resistor_world, screen_tech, core_components = sup_vids # Products prod_vids = db.bulk_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 db.bulk_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 db.bulk_insert_edges('ASSEMBLED_FROM', [ (tablet, mbx1, {}), (tablet, scr_lcd, {}), (tablet, bat4000, {}), (tablet, proc_arm, {}), ]) # Supply relationships db.bulk_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, {}), ]) db.flush() print("Data ingested")

Data ingested

3. BOM Explosion¶

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

In [5]:

query_bom = """
    MATCH (defective:Part {sku: 'RES-10K'})
    MATCH (product:Product)-[:ASSEMBLED_FROM*1..5]->(defective)
    RETURN product.name AS name, product.price AS price
    ORDER BY product.price DESC
"""
results = db.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)}'

query_bom = """ MATCH (defective:Part {sku: 'RES-10K'}) MATCH (product:Product)-[:ASSEMBLED_FROM*1..5]->(defective) RETURN product.name AS name, product.price AS price ORDER BY product.price DESC """ results = db.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)

4. Full BOM Listing¶

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

In [6]:

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 = db.query(query_parts)
print('Smartphone X BOM:')
for r in results:
    print(f"  {r['part_name']} ({r['sku']}): ${r['cost']}")

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 = db.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.

In [7]:

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 = db.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)}'

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 = db.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?

In [8]:

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,
           COUNT(DISTINCT part) AS parts_supplied
    ORDER BY products_at_risk DESC, parts_supplied DESC
"""
results = db.query(query_risk)
print('Supplier risk analysis:')
for r in results:
    print(f"  {r['supplier']}: {r['products_at_risk']} product(s), {r['parts_supplied']} part(s)")
top = results[0]
assert top['supplier'] == 'CoreComponents', f"Expected CoreComponents, got {top['supplier']}"
assert top['products_at_risk'] == 2, f"Expected 2, got {top['products_at_risk']}"

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, COUNT(DISTINCT part) AS parts_supplied ORDER BY products_at_risk DESC, parts_supplied DESC """ results = db.query(query_risk) print('Supplier risk analysis:') for r in results: print(f" {r['supplier']}: {r['products_at_risk']} product(s), {r['parts_supplied']} part(s)") top = results[0] assert top['supplier'] == 'CoreComponents', f"Expected CoreComponents, got {top['supplier']}" assert top['products_at_risk'] == 2, f"Expected 2, got {top['products_at_risk']}"

Supplier risk analysis:
  CoreComponents: 2 product(s), 3 part(s)
  ScreenTech: 2 product(s), 2 part(s)
  ResistorWorld: 2 product(s), 2 part(s)