Supply Chain Management with uni-pydantic¶

BOM explosion, cost rollup, and supplier risk analysis using Pydantic models.

In [1]:

import os
import shutil
import tempfile

import uni_db
from uni_pydantic import UniNode, UniEdge, UniSession, Field, Relationship

import os import shutil import tempfile import uni_db from uni_pydantic import UniNode, UniEdge, UniSession, Field, Relationship

1. Define Models¶

Parts, Suppliers, and Products with assembly relationships using type-safe Pydantic models.

In [2]:

class Part(UniNode):
    """A component part in the supply chain."""
    __label__ = "Part"
    
    name: str
    sku: str = Field(index="hash", unique=True)
    cost: float
    
    # Relationships
    used_in: list["Part"] = Relationship("ASSEMBLED_FROM", direction="incoming")
    components: list["Part"] = Relationship("ASSEMBLED_FROM", direction="outgoing")
    suppliers: list["Supplier"] = Relationship("SUPPLIED_BY", direction="outgoing")


class Supplier(UniNode):
    """A supplier of parts."""
    __label__ = "Supplier"
    
    name: str = Field(index="btree")
    
    # Relationships
    supplies: list[Part] = Relationship("SUPPLIED_BY", direction="incoming")


class Product(UniNode):
    """A finished product assembled from parts."""
    __label__ = "Product"
    
    name: str = Field(index="btree")
    price: float
    
    # Relationships
    components: list[Part] = Relationship("ASSEMBLED_FROM", direction="outgoing")


class AssembledFrom(UniEdge):
    """Edge representing assembly relationship."""
    __edge_type__ = "ASSEMBLED_FROM"
    __from__ = (Product, Part)
    __to__ = Part


class SuppliedBy(UniEdge):
    """Edge representing supplier relationship."""
    __edge_type__ = "SUPPLIED_BY"
    __from__ = Part
    __to__ = Supplier

class Part(UniNode): """A component part in the supply chain.""" __label__ = "Part" name: str sku: str = Field(index="hash", unique=True) cost: float # Relationships used_in: list["Part"] = Relationship("ASSEMBLED_FROM", direction="incoming") components: list["Part"] = Relationship("ASSEMBLED_FROM", direction="outgoing") suppliers: list["Supplier"] = Relationship("SUPPLIED_BY", direction="outgoing") class Supplier(UniNode): """A supplier of parts.""" __label__ = "Supplier" name: str = Field(index="btree") # Relationships supplies: list[Part] = Relationship("SUPPLIED_BY", direction="incoming") class Product(UniNode): """A finished product assembled from parts.""" __label__ = "Product" name: str = Field(index="btree") price: float # Relationships components: list[Part] = Relationship("ASSEMBLED_FROM", direction="outgoing") class AssembledFrom(UniEdge): """Edge representing assembly relationship.""" __edge_type__ = "ASSEMBLED_FROM" __from__ = (Product, Part) __to__ = Part class SuppliedBy(UniEdge): """Edge representing supplier relationship.""" __edge_type__ = "SUPPLIED_BY" __from__ = Part __to__ = Supplier

2. Setup Database and Session¶

In [3]:

db_path = os.path.join(tempfile.gettempdir(), "supply_chain_pydantic_db")
if os.path.exists(db_path):
    shutil.rmtree(db_path)
db = uni_db.Database(db_path)

# Create session and register models
session = UniSession(db)
session.register(Part, Supplier, Product, AssembledFrom, SuppliedBy)
session.sync_schema()

print(f"Opened database at {db_path}")

db_path = os.path.join(tempfile.gettempdir(), "supply_chain_pydantic_db") if os.path.exists(db_path): shutil.rmtree(db_path) db = uni_db.Database(db_path) # Create session and register models session = UniSession(db) session.register(Part, Supplier, Product, AssembledFrom, SuppliedBy) session.sync_schema() print(f"Opened database at {db_path}")

Opened database at /tmp/supply_chain_pydantic_db

3. Create Data¶

Two products sharing common parts, supplied by multiple vendors.

In [4]:

# 7 parts: resistors, capacitors, boards, screens, battery, processor
res10k   = Part(name="Resistor 10K",    sku="RES-10K",   cost=0.05)
cap100uf = Part(name="Capacitor 100uF", sku="CAP-100UF", cost=0.08)
mbx1     = Part(name="Motherboard X1",  sku="MB-X1",     cost=50.0)
scr_oled = Part(name="OLED Screen",     sku="SCR-OLED",  cost=30.0)
bat4000  = Part(name="Battery 4000mAh", sku="BAT-4000",  cost=15.0)
proc_arm = Part(name="ARM Processor",   sku="PROC-ARM",  cost=80.0)
scr_lcd  = Part(name="LCD Screen",      sku="SCR-LCD",   cost=20.0)

# 3 suppliers
resistor_world  = Supplier(name="ResistorWorld")
screen_tech     = Supplier(name="ScreenTech")
core_components = Supplier(name="CoreComponents")

# 2 products
smartphone = Product(name="Smartphone X",  price=599.0)
tablet     = Product(name="TabletPro 10",  price=799.0)

session.add_all([
    res10k, cap100uf, mbx1, scr_oled, bat4000, proc_arm, scr_lcd,
    resistor_world, screen_tech, core_components,
    smartphone, tablet,
])
session.commit()
print("Nodes created")

# 7 parts: resistors, capacitors, boards, screens, battery, processor res10k = Part(name="Resistor 10K", sku="RES-10K", cost=0.05) cap100uf = Part(name="Capacitor 100uF", sku="CAP-100UF", cost=0.08) mbx1 = Part(name="Motherboard X1", sku="MB-X1", cost=50.0) scr_oled = Part(name="OLED Screen", sku="SCR-OLED", cost=30.0) bat4000 = Part(name="Battery 4000mAh", sku="BAT-4000", cost=15.0) proc_arm = Part(name="ARM Processor", sku="PROC-ARM", cost=80.0) scr_lcd = Part(name="LCD Screen", sku="SCR-LCD", cost=20.0) # 3 suppliers resistor_world = Supplier(name="ResistorWorld") screen_tech = Supplier(name="ScreenTech") core_components = Supplier(name="CoreComponents") # 2 products smartphone = Product(name="Smartphone X", price=599.0) tablet = Product(name="TabletPro 10", price=799.0) session.add_all([ res10k, cap100uf, mbx1, scr_oled, bat4000, proc_arm, scr_lcd, resistor_world, screen_tech, core_components, smartphone, tablet, ]) session.commit() print("Nodes created")

Nodes created

In [5]:

# Smartphone X assembly
session.create_edge(smartphone, "ASSEMBLED_FROM", mbx1)
session.create_edge(smartphone, "ASSEMBLED_FROM", scr_oled)
session.create_edge(smartphone, "ASSEMBLED_FROM", bat4000)
session.create_edge(smartphone, "ASSEMBLED_FROM", proc_arm)
session.create_edge(mbx1, "ASSEMBLED_FROM", res10k)
session.create_edge(mbx1, "ASSEMBLED_FROM", cap100uf)

# TabletPro 10 assembly (shares mbx1, bat4000, proc_arm)
session.create_edge(tablet, "ASSEMBLED_FROM", mbx1)
session.create_edge(tablet, "ASSEMBLED_FROM", scr_lcd)
session.create_edge(tablet, "ASSEMBLED_FROM", bat4000)
session.create_edge(tablet, "ASSEMBLED_FROM", proc_arm)

# Supply relationships
session.create_edge(res10k,   "SUPPLIED_BY", resistor_world)
session.create_edge(cap100uf, "SUPPLIED_BY", resistor_world)
session.create_edge(scr_oled, "SUPPLIED_BY", screen_tech)
session.create_edge(scr_lcd,  "SUPPLIED_BY", screen_tech)
session.create_edge(mbx1,     "SUPPLIED_BY", core_components)
session.create_edge(bat4000,  "SUPPLIED_BY", core_components)
session.create_edge(proc_arm, "SUPPLIED_BY", core_components)

session.commit()
print("Data ingested")

# Smartphone X assembly session.create_edge(smartphone, "ASSEMBLED_FROM", mbx1) session.create_edge(smartphone, "ASSEMBLED_FROM", scr_oled) session.create_edge(smartphone, "ASSEMBLED_FROM", bat4000) session.create_edge(smartphone, "ASSEMBLED_FROM", proc_arm) session.create_edge(mbx1, "ASSEMBLED_FROM", res10k) session.create_edge(mbx1, "ASSEMBLED_FROM", cap100uf) # TabletPro 10 assembly (shares mbx1, bat4000, proc_arm) session.create_edge(tablet, "ASSEMBLED_FROM", mbx1) session.create_edge(tablet, "ASSEMBLED_FROM", scr_lcd) session.create_edge(tablet, "ASSEMBLED_FROM", bat4000) session.create_edge(tablet, "ASSEMBLED_FROM", proc_arm) # Supply relationships session.create_edge(res10k, "SUPPLIED_BY", resistor_world) session.create_edge(cap100uf, "SUPPLIED_BY", resistor_world) session.create_edge(scr_oled, "SUPPLIED_BY", screen_tech) session.create_edge(scr_lcd, "SUPPLIED_BY", screen_tech) session.create_edge(mbx1, "SUPPLIED_BY", core_components) session.create_edge(bat4000, "SUPPLIED_BY", core_components) session.create_edge(proc_arm, "SUPPLIED_BY", core_components) session.commit() print("Data ingested")

Data ingested

4. BOM Explosion¶

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

In [6]:

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 = session.cypher(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 = session.cypher(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)

5. Full BOM Listing¶

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

In [7]:

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.cypher(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 = session.cypher(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

6. Cost Rollup¶

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

In [8]:

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.cypher(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 = session.cypher(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

7. Supply Chain Risk¶

Which supplier is critical to the most products?

In [9]:

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 = session.cypher(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 = session.cypher(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)

8. Query Builder Demo¶

Using the type-safe query builder to find expensive parts.

In [10]:

# Find all expensive parts using query builder
expensive_parts = (
    session.query(Part)
    .filter(Part.cost >= 20.0)
    .order_by(Part.cost, descending=True)
    .all()
)

print("Expensive Parts (>=$20):")
for part in expensive_parts:
    print(f"  - {part.name} ({part.sku}): ${part.cost:,.2f}")

# Find all expensive parts using query builder expensive_parts = ( session.query(Part) .filter(Part.cost >= 20.0) .order_by(Part.cost, descending=True) .all() ) print("Expensive Parts (>=$20):") for part in expensive_parts: print(f" - {part.name} ({part.sku}): ${part.cost:,.2f}")

Expensive Parts (>=$20):
  - ARM Processor (PROC-ARM): $80.00
  - Motherboard X1 (MB-X1): $50.00
  - OLED Screen (SCR-OLED): $30.00
  - LCD Screen (SCR-LCD): $20.00