Supply Chain Management with Uni¶

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

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¶

We define Parts, Suppliers, and Products, along with relationships for Assembly and Supply.

In [3]:

db.create_label("Part")
db.create_label("Supplier")
db.create_label("Product")

db.create_edge_type("ASSEMBLED_FROM", ["Product", "Part"], ["Part"])
db.create_edge_type("SUPPLIED_BY", ["Part"], ["Supplier"])

db.add_property("Part", "sku", "string", False)
db.add_property("Part", "cost", "float64", False)
db.add_property("Product", "name", "string", False)
db.add_property("Product", "price", "float64", False)

db.create_scalar_index("Part", "sku", "btree")

db.create_label("Part") db.create_label("Supplier") db.create_label("Product") db.create_edge_type("ASSEMBLED_FROM", ["Product", "Part"], ["Part"]) db.create_edge_type("SUPPLIED_BY", ["Part"], ["Supplier"]) db.add_property("Part", "sku", "string", False) db.add_property("Part", "cost", "float64", False) db.add_property("Product", "name", "string", False) db.add_property("Product", "price", "float64", False) db.create_scalar_index("Part", "sku", "btree")

2. Ingest Data¶

We insert parts and products using bulk insertion for performance.

In [4]:

p1_props = {"sku": "RES-10K", "cost": 0.05, "_doc": {"type": "resistor", "compliance": ["RoHS"]}}
p2_props = {"sku": "MB-X1", "cost": 50.0}
p3_props = {"sku": "SCR-OLED", "cost": 30.0}

vids = db.bulk_insert_vertices("Part", [p1_props, p2_props, p3_props])
p1, p2, p3 = vids

prod_props = {"name": "Smartphone X", "price": 500.0}
phone_vids = db.bulk_insert_vertices("Product", [prod_props])
phone = phone_vids[0]

db.bulk_insert_edges("ASSEMBLED_FROM", [(phone, p2, {}), (phone, p3, {}), (p2, p1, {})])

db.flush()

p1_props = {"sku": "RES-10K", "cost": 0.05, "_doc": {"type": "resistor", "compliance": ["RoHS"]}} p2_props = {"sku": "MB-X1", "cost": 50.0} p3_props = {"sku": "SCR-OLED", "cost": 30.0} vids = db.bulk_insert_vertices("Part", [p1_props, p2_props, p3_props]) p1, p2, p3 = vids prod_props = {"name": "Smartphone X", "price": 500.0} phone_vids = db.bulk_insert_vertices("Product", [prod_props]) phone = phone_vids[0] db.bulk_insert_edges("ASSEMBLED_FROM", [(phone, p2, {}), (phone, p3, {}), (p2, p1, {})]) db.flush()

3. BOM Explosion¶

Find all products that contain a specific defective part (RES-10K), traversing up the assembly hierarchy.

In [5]:

# Warm-up to ensure adjacency is loaded
db.query("MATCH (a:Part)-[:ASSEMBLED_FROM]->(b:Part) RETURN a.sku")

query = "MATCH (defective:Part {sku: 'RES-10K'}) MATCH (product:Product)-[:ASSEMBLED_FROM*1..5]->(defective) RETURN product.name as name, product.price as price"
results = db.query(query)
print("Products affected:")
for r in results:
    print(r)

# Warm-up to ensure adjacency is loaded db.query("MATCH (a:Part)-[:ASSEMBLED_FROM]->(b:Part) RETURN a.sku") query = "MATCH (defective:Part {sku: 'RES-10K'}) MATCH (product:Product)-[:ASSEMBLED_FROM*1..5]->(defective) RETURN product.name as name, product.price as price" results = db.query(query) print("Products affected:") for r in results: print(r)

Products affected:
{'name': 'Smartphone X', 'price': 500.0}
{'price': 500.0, 'name': 'Smartphone X'}
{'name': 'Smartphone X', 'price': 500.0}

4. Cost Rollup¶

Calculate the total cost of parts for a product by traversing down the assembly tree.

In [6]:

query_cost = "MATCH (p:Product {name: 'Smartphone X'}) MATCH (p)-[:ASSEMBLED_FROM*1..5]->(part:Part) RETURN SUM(part.cost) AS total_bom_cost"
results_cost = db.query(query_cost)
print(f"Total BOM Cost: {results_cost[0]['total_bom_cost']}")

query_cost = "MATCH (p:Product {name: 'Smartphone X'}) MATCH (p)-[:ASSEMBLED_FROM*1..5]->(part:Part) RETURN SUM(part.cost) AS total_bom_cost" results_cost = db.query(query_cost) print(f"Total BOM Cost: {results_cost[0]['total_bom_cost']}")

Total BOM Cost: 80.05

DEBUG 2: DataFusion execution failed (falling back to execute_subplan): Schema error: No field named "part.cost". Valid fields are "p._vid", "p.name", _target_vid, _hop_count.