Supply Chain Management with uni-pydantic¶

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

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.

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¶

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

# 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

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

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

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.

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

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?

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.

# 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