← Back to all projects

LEARN CPQ CONFIGURE PRICE QUOTE DEEP DIVE

Learn CPQ (Configure, Price, Quote): Build Enterprise Sales Software from Scratch

Goal: Deeply understand CPQ systems—from product catalogs and configuration engines through pricing algorithms and quote generation, to approval workflows and CRM/ERP integration.


Why CPQ Matters

Every B2B company with complex products faces the same challenge: how do you let sales teams quickly configure products, apply correct pricing, and generate professional quotes? CPQ software solves this, and it’s a multi-billion dollar industry.

Understanding CPQ means understanding:

  • How enterprise software handles complex business rules
  • How to model configurable products with constraints
  • How pricing engines handle discounts, tiers, and bundles
  • How approval workflows enforce business policies
  • How to generate professional documents from data
  • How to integrate with CRM (Salesforce) and ERP systems

After completing these projects, you will:

  • Build complete CPQ systems from scratch
  • Design product configuration engines with constraint satisfaction
  • Implement flexible pricing engines with multiple strategies
  • Create quote document generators with templates
  • Build approval workflows with routing rules
  • Integrate with external systems via APIs
  • Understand why companies pay millions for CPQ solutions

Core Concept Analysis

What is CPQ?

CPQ stands for Configure, Price, Quote—the three-step process sales teams follow:

┌─────────────────────────────────────────────────────────────────────────────┐
│                           THE CPQ PROCESS                                    │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│   1. CONFIGURE                2. PRICE                 3. QUOTE             │
│   ┌──────────────┐           ┌──────────────┐        ┌──────────────┐       │
│   │ Product      │           │ Calculate    │        │ Generate     │       │
│   │ Selection    │ ────────► │ Pricing      │ ─────► │ Document     │       │
│   │ & Options    │           │ & Discounts  │        │ & Send       │       │
│   └──────────────┘           └──────────────┘        └──────────────┘       │
│                                                                              │
│   • Select products          • Base prices           • Professional PDF     │
│   • Choose options           • Volume discounts      • Terms & conditions   │
│   • Validate configs         • Bundle pricing        • E-signature          │
│   • Handle dependencies      • Approval routing      • Track & follow up    │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

The Business Problem

Without CPQ, sales teams face:

  • Configuration errors: Selling incompatible options
  • Pricing mistakes: Wrong discounts, missing fees
  • Slow quotes: Days instead of minutes
  • Inconsistent proposals: Every rep does it differently
  • Compliance issues: Unapproved discounts, missing terms

CPQ System Architecture

┌─────────────────────────────────────────────────────────────────────────────┐
│                         CPQ SYSTEM ARCHITECTURE                              │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  ┌──────────────────────────────────────────────────────────────────────┐   │
│  │                        PRESENTATION LAYER                             │   │
│  │  ┌────────────┐  ┌────────────┐  ┌────────────┐  ┌────────────┐      │   │
│  │  │ Sales Rep  │  │ Customer   │  │ Mobile     │  │ Partner    │      │   │
│  │  │ UI         │  │ Self-Serve │  │ App        │  │ Portal     │      │   │
│  │  └────────────┘  └────────────┘  └────────────┘  └────────────┘      │   │
│  └──────────────────────────────────────────────────────────────────────┘   │
│                                      │                                       │
│  ┌──────────────────────────────────┴───────────────────────────────────┐   │
│  │                         CPQ ENGINE                                    │   │
│  │  ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐         │   │
│  │  │  Configuration  │ │    Pricing      │ │    Quoting      │         │   │
│  │  │     Engine      │ │    Engine       │ │    Engine       │         │   │
│  │  │                 │ │                 │ │                 │         │   │
│  │  │ • Rules Engine  │ │ • Price Lists   │ │ • Templates     │         │   │
│  │  │ • Constraints   │ │ • Discounts     │ │ • PDF Generator │         │   │
│  │  │ • Validation    │ │ • Tiers/Volume  │ │ • E-Signature   │         │   │
│  │  └─────────────────┘ └─────────────────┘ └─────────────────┘         │   │
│  │                                                                       │   │
│  │  ┌─────────────────┐ ┌─────────────────┐                              │   │
│  │  │    Approval     │ │   Guided        │                              │   │
│  │  │    Workflow     │ │   Selling       │                              │   │
│  │  └─────────────────┘ └─────────────────┘                              │   │
│  └──────────────────────────────────────────────────────────────────────┘   │
│                                      │                                       │
│  ┌──────────────────────────────────┴───────────────────────────────────┐   │
│  │                         DATA LAYER                                    │   │
│  │  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐     │   │
│  │  │  Product    │ │   Price     │ │   Quote     │ │  Customer   │     │   │
│  │  │  Catalog    │ │   Books     │ │   History   │ │   Data      │     │   │
│  │  └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘     │   │
│  └──────────────────────────────────────────────────────────────────────┘   │
│                                      │                                       │
│  ┌──────────────────────────────────┴───────────────────────────────────┐   │
│  │                      INTEGRATIONS                                     │   │
│  │  ┌────────┐  ┌────────┐  ┌────────┐  ┌────────┐  ┌────────┐          │   │
│  │  │  CRM   │  │  ERP   │  │ E-Sign │  │ Payment│  │ DocGen │          │   │
│  │  │Salesforce│ │SAP/Oracle│ │DocuSign│ │ Stripe │  │        │          │   │
│  │  └────────┘  └────────┘  └────────┘  └────────┘  └────────┘          │   │
│  └──────────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────────┘

Key Concepts

1. Product Configuration

Products in CPQ have complex structures:

PRODUCT: Enterprise Software License
├── Base Product (required)
│   └── $10,000/year base
├── User Tiers (choose one)
│   ├── Up to 100 users: +$5,000
│   ├── 101-500 users: +$15,000
│   └── 501-1000 users: +$30,000
├── Modules (choose multiple)
│   ├── Analytics: +$3,000/year
│   ├── API Access: +$5,000/year
│   ├── SSO Integration: +$2,000/year (requires: API Access)
│   └── Advanced Security: +$4,000/year
├── Support Level (choose one)
│   ├── Standard: included
│   ├── Premium 8x5: +$2,000
│   └── Premium 24x7: +$5,000
└── Professional Services (optional)
    ├── Implementation: $15,000 one-time
    └── Training: $500/user

RULES:
- If users > 500: Premium support required
- SSO requires API Access
- Advanced Security + 24x7 Support = 10% discount

2. Configuration Rules Types

Rule Type Description Example
Inclusion Automatically add product “Enterprise → include Support”
Exclusion Prevent combination “Basic plan excludes API”
Dependency Require another product “SSO requires API Access”
Validation Check conditions “Max 1000 users per license”
Recommendation Suggest products “Customers who bought X also bought Y”

3. Pricing Strategies

PRICING COMPONENTS:
┌─────────────────────────────────────────────────────────────────┐
│                                                                  │
│   List Price      $100,000                                       │
│   ─────────────────────────                                      │
│   Volume Discount  -10% (>$50K order)     -$10,000              │
│   Customer Discount -5% (Gold partner)     -$4,500              │
│   Bundle Discount   -8% (3+ products)      -$6,840              │
│   Promotional       -$5,000 (Q4 special)   -$5,000              │
│   ─────────────────────────                                      │
│   Net Price        $73,660                                       │
│   ─────────────────────────                                      │
│   Partner Margin   +15%                    +$11,049             │
│   ─────────────────────────                                      │
│   Final Price      $84,709                                       │
│                                                                  │
│   APPROVAL NEEDED: Discount > 20% (current: 26.34%)             │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Pricing Models:

  • Flat pricing: Fixed price per unit
  • Tiered pricing: Different price at each tier (first 10 = $100, next 10 = $90)
  • Volume pricing: All units at volume price once threshold hit
  • Usage-based: Pay for what you use
  • Subscription: Recurring fees with terms

4. Approval Workflows

                    Quote Created
                         │
                         ▼
              ┌──────────────────┐
              │ Check Conditions │
              └────────┬─────────┘
                       │
         ┌─────────────┼─────────────┐
         │             │             │
         ▼             ▼             ▼
    Discount      Deal Size      Special
    > 15%?        > $100K?       Terms?
         │             │             │
         ▼             ▼             ▼
    ┌────────┐   ┌────────┐   ┌────────┐
    │Sales   │   │Regional│   │Legal   │
    │Manager │   │Director│   │Review  │
    └───┬────┘   └───┬────┘   └───┬────┘
        │            │            │
        └────────────┴────────────┘
                     │
                     ▼
              ┌──────────────┐
              │ All Approved │
              └──────┬───────┘
                     │
                     ▼
              Quote Finalized

5. Quote Document Structure

┌──────────────────────────────────────────────────────────────────┐
│                                                                   │
│   [COMPANY LOGO]                    Quote #: Q-2024-0001234      │
│                                     Date: December 21, 2024       │
│                                     Valid Until: January 20, 2025 │
│                                                                   │
│   PREPARED FOR:                     PREPARED BY:                  │
│   Acme Corporation                  Jane Smith                    │
│   123 Business Ave                  Senior Account Executive      │
│   New York, NY 10001                jane.smith@ourcompany.com    │
│                                                                   │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│   QUOTE SUMMARY                                                   │
│   ─────────────────────────────────────────────────────────────  │
│                                                                   │
│   Product                    Qty    Unit Price    Total          │
│   ─────────────────────────────────────────────────────────────  │
│   Enterprise License          1     $10,000.00    $10,000.00     │
│   User Tier (101-500)         1     $15,000.00    $15,000.00     │
│   Analytics Module            1      $3,000.00     $3,000.00     │
│   API Access Module           1      $5,000.00     $5,000.00     │
│   Premium Support 24x7        1      $5,000.00     $5,000.00     │
│   ─────────────────────────────────────────────────────────────  │
│   Subtotal                                        $38,000.00     │
│   Volume Discount (10%)                           -$3,800.00     │
│   ─────────────────────────────────────────────────────────────  │
│   TOTAL                                           $34,200.00     │
│                                                                   │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│   TERMS & CONDITIONS                                              │
│   • Payment due within 30 days of invoice                        │
│   • Annual subscription, auto-renews                             │
│   • Subject to Master Services Agreement                         │
│                                                                   │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│   ACCEPTANCE                                                      │
│                                                                   │
│   _________________________    _____________                      │
│   Signature                    Date                               │
│                                                                   │
│   _________________________                                       │
│   Printed Name                                                    │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

Project List

Projects are ordered from foundational concepts to full CPQ implementation.


Project 1: Product Catalog with Configurable Products

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: Python (Django/FastAPI)
  • Alternative Programming Languages: TypeScript (Node.js), Java (Spring Boot), C# (.NET)
  • Coolness Level: Level 2: Practical but Forgettable
  • Business Potential: 3. The “Service & Support” Model
  • Difficulty: Level 2: Intermediate
  • Knowledge Area: Data Modeling / Product Information Management
  • Software or Tool: PostgreSQL, Redis
  • Main Book: “Domain-Driven Design” by Eric Evans

What you’ll build: A flexible product catalog that handles simple products, configurable products with options, bundles, and product relationships (requires, excludes, recommends).

Why it teaches CPQ foundations: The product catalog is the foundation of any CPQ system. Without a proper data model for configurable products, you can’t build configuration rules or accurate pricing.

Core challenges you’ll face:

  • Modeling product variants → maps to EAV patterns, attribute inheritance
  • Handling product relationships → maps to graph structures, dependency management
  • Supporting multiple product types → maps to polymorphism, type hierarchies
  • Version control for products → maps to temporal data, audit trails

Key Concepts:

  • Product Data Modeling: commercetools Product Modeling
  • EAV Pattern: “Patterns of Enterprise Application Architecture” - Martin Fowler
  • Domain Modeling: “Domain-Driven Design” Chapter 5 - Eric Evans

Difficulty: Intermediate Time estimate: 1-2 weeks Prerequisites: Understanding of relational databases, REST APIs

Real world outcome:

# Your Product Catalog API

# Create a configurable product
product = catalog.create_product({
    "name": "Enterprise Software License",
    "type": "configurable",
    "base_price": 10000,
    "attributes": [
        {"name": "user_tier", "type": "select", "required": True,
         "options": ["small", "medium", "large"]},
        {"name": "modules", "type": "multiselect",
         "options": ["analytics", "api", "sso", "security"]},
        {"name": "support_level", "type": "select",
         "options": ["standard", "premium", "24x7"]}
    ],
    "relationships": [
        {"type": "requires", "source": "sso", "target": "api"},
        {"type": "excludes", "source": "basic_plan", "target": "api"},
        {"type": "recommends", "source": "large_tier", "target": "24x7"}
    ]
})

# Query products
products = catalog.search(
    category="software",
    attributes={"user_tier": "large"},
    include_relationships=True
)

Implementation Hints:

  1. Core Data Model:
    -- Products table (base)
    CREATE TABLE products (
        id UUID PRIMARY KEY,
        sku VARCHAR(50) UNIQUE,
        name VARCHAR(255) NOT NULL,
        type VARCHAR(50) NOT NULL, -- simple, configurable, bundle
        status VARCHAR(20) DEFAULT 'active',
        base_price DECIMAL(12,2),
        created_at TIMESTAMP DEFAULT NOW(),
        updated_at TIMESTAMP DEFAULT NOW()
    );
    
    -- Product attributes definition
    CREATE TABLE product_attributes (
        id UUID PRIMARY KEY,
        product_id UUID REFERENCES products(id),
        name VARCHAR(100) NOT NULL,
        type VARCHAR(50) NOT NULL, -- text, number, select, multiselect
        required BOOLEAN DEFAULT FALSE,
        default_value JSONB
    );
    
    -- Attribute options (for select/multiselect)
    CREATE TABLE attribute_options (
        id UUID PRIMARY KEY,
        attribute_id UUID REFERENCES product_attributes(id),
        value VARCHAR(255) NOT NULL,
        label VARCHAR(255),
        price_modifier DECIMAL(12,2) DEFAULT 0,
        sort_order INTEGER DEFAULT 0
    );
    
    -- Product relationships
    CREATE TABLE product_relationships (
        id UUID PRIMARY KEY,
        source_product_id UUID REFERENCES products(id),
        target_product_id UUID REFERENCES products(id),
        relationship_type VARCHAR(50), -- requires, excludes, recommends, includes
        source_condition JSONB, -- When source has these attributes...
        target_condition JSONB  -- ...apply this to target
    );
    
    -- Bundle components
    CREATE TABLE bundle_components (
        id UUID PRIMARY KEY,
        bundle_id UUID REFERENCES products(id),
        component_id UUID REFERENCES products(id),
        quantity INTEGER DEFAULT 1,
        required BOOLEAN DEFAULT TRUE,
        discount_percent DECIMAL(5,2) DEFAULT 0
    );
    
  2. Product Service:
    class ProductCatalog:
        def create_product(self, data: dict) -> Product:
            product = Product(
                sku=data['sku'],
                name=data['name'],
                type=ProductType(data['type']),
                base_price=Decimal(data.get('base_price', 0))
            )
    
            # Add attributes
            for attr_data in data.get('attributes', []):
                attribute = ProductAttribute(
                    name=attr_data['name'],
                    type=AttributeType(attr_data['type']),
                    required=attr_data.get('required', False)
                )
                for opt in attr_data.get('options', []):
                    attribute.options.append(AttributeOption(
                        value=opt['value'] if isinstance(opt, dict) else opt,
                        price_modifier=opt.get('price_modifier', 0) if isinstance(opt, dict) else 0
                    ))
                product.attributes.append(attribute)
    
            # Add relationships
            for rel_data in data.get('relationships', []):
                product.relationships.append(ProductRelationship(
                    type=RelationshipType(rel_data['type']),
                    target_id=rel_data['target_id'],
                    condition=rel_data.get('condition')
                ))
    
            self.db.add(product)
            self.db.commit()
            return product
    
        def get_related_products(self, product_id: UUID, rel_type: str = None):
            query = self.db.query(ProductRelationship).filter(
                ProductRelationship.source_product_id == product_id
            )
            if rel_type:
                query = query.filter(ProductRelationship.type == rel_type)
            return query.all()
    
  3. Version Control (for product changes):
    class ProductVersion:
        def save_version(self, product: Product):
            version = ProductVersionRecord(
                product_id=product.id,
                version_number=self.get_next_version(product.id),
                data=product.to_dict(),
                created_at=datetime.now(),
                created_by=self.current_user
            )
            self.db.add(version)
    
        def get_product_at_version(self, product_id: UUID, version: int):
            record = self.db.query(ProductVersionRecord).filter(
                ProductVersionRecord.product_id == product_id,
                ProductVersionRecord.version_number == version
            ).first()
            return Product.from_dict(record.data)
    

Learning milestones:

  1. Simple products work → You understand basic modeling
  2. Configurable products with options → You understand attributes
  3. Relationships work → You understand product dependencies
  4. Bundles work → You understand composite products
  5. Version history works → You understand temporal data

Project 2: Configuration Rules Engine

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: Python
  • Alternative Programming Languages: TypeScript, Java, Rust
  • Coolness Level: Level 4: Hardcore Tech Flex
  • Business Potential: 4. The “Open Core” Infrastructure
  • Difficulty: Level 3: Advanced
  • Knowledge Area: Rules Engines / Constraint Satisfaction
  • Software or Tool: Custom rules engine or Drools/Easy Rules
  • Main Book: “Production Systems in Artificial Intelligence” - various papers

What you’ll build: A rules engine that validates product configurations, enforces dependencies, suggests products, and prevents invalid combinations—the “brain” of the CPQ system.

Why it teaches configuration logic: Configuration is where CPQ gets complex. Rule-based systems vs constraint satisfaction, forward vs backward chaining, and handling combinatorial explosion are all real engineering challenges.

Core challenges you’ll face:

  • Rule representation → maps to DSL design, JSON rules, database rules
  • Rule evaluation order → maps to priority, conflict resolution
  • Dependency resolution → maps to graph traversal, cycle detection
  • Performance with many rules → maps to Rete algorithm, indexing

Key Concepts:

Difficulty: Advanced Time estimate: 2-3 weeks Prerequisites: Project 1 completed, understanding of algorithms

Real world outcome:

# Define configuration rules
rules_engine.add_rule(
    name="SSO requires API",
    type="dependency",
    condition={"selected": "sso_module"},
    action={"require": "api_module"},
    message="SSO Integration requires API Access"
)

rules_engine.add_rule(
    name="Large tier needs premium support",
    type="validation",
    condition={"user_tier": "large"},
    action={"require": {"support_level": ["premium", "24x7"]}},
    message="Enterprise tier (500+ users) requires Premium support"
)

rules_engine.add_rule(
    name="Bundle discount",
    type="recommendation",
    condition={"selected_count": {"modules": {"gte": 3}}},
    action={"suggest": "bundle_discount", "discount": 0.1},
    message="Add 1 more module to qualify for 10% bundle discount!"
)

# Validate a configuration
config = {
    "product_id": "enterprise-license",
    "selections": {
        "user_tier": "large",
        "modules": ["analytics", "sso"],
        "support_level": "standard"
    }
}

result = rules_engine.validate(config)
# Result:
# {
#     "valid": False,
#     "errors": [
#         {"rule": "SSO requires API", "message": "SSO Integration requires API Access"},
#         {"rule": "Large tier needs premium support", "message": "..."}
#     ],
#     "warnings": [],
#     "suggestions": [
#         {"rule": "Bundle discount", "message": "Add 1 more module..."}
#     ],
#     "auto_additions": [
#         {"product": "api_module", "reason": "Required by SSO"}
#     ]
# }

Implementation Hints:

  1. Rule Data Model:
    from enum import Enum
    from dataclasses import dataclass
    from typing import Dict, List, Any, Optional
    
    class RuleType(Enum):
        VALIDATION = "validation"      # Check if config is valid
        DEPENDENCY = "dependency"      # A requires B
        EXCLUSION = "exclusion"        # A excludes B
        INCLUSION = "inclusion"        # A auto-adds B
        RECOMMENDATION = "recommendation"  # Suggest B when A
    
    class ActionType(Enum):
        REQUIRE = "require"
        EXCLUDE = "exclude"
        ADD = "add"
        REMOVE = "remove"
        SUGGEST = "suggest"
        WARN = "warn"
        ERROR = "error"
    
    @dataclass
    class Rule:
        id: str
        name: str
        type: RuleType
        condition: Dict[str, Any]  # When this is true...
        action: Dict[str, Any]     # ...do this
        priority: int = 100        # Higher = runs first
        enabled: bool = True
        message: str = ""
    
    @dataclass
    class RuleResult:
        rule: Rule
        matched: bool
        action_type: ActionType
        details: Dict[str, Any]
    
  2. Condition Evaluator:
    class ConditionEvaluator:
        def evaluate(self, condition: Dict, context: Dict) -> bool:
            """Evaluate a condition against a configuration context."""
            if not condition:
                return True
    
            for key, expected in condition.items():
                if key == "$and":
                    return all(self.evaluate(c, context) for c in expected)
                elif key == "$or":
                    return any(self.evaluate(c, context) for c in expected)
                elif key == "$not":
                    return not self.evaluate(expected, context)
                elif key == "selected":
                    # Check if product/option is selected
                    return self._is_selected(expected, context)
                elif key == "selected_count":
                    # Check count of selections
                    return self._check_count(expected, context)
                else:
                    # Direct attribute comparison
                    return self._compare(key, expected, context)
    
            return True
    
        def _is_selected(self, item: str, context: Dict) -> bool:
            selections = context.get('selections', {})
            for category, selected in selections.items():
                if isinstance(selected, list) and item in selected:
                    return True
                elif selected == item:
                    return True
            return False
    
        def _check_count(self, spec: Dict, context: Dict) -> bool:
            for category, comparison in spec.items():
                actual_count = len(context.get('selections', {}).get(category, []))
                for op, value in comparison.items():
                    if op == "gte" and actual_count < value:
                        return False
                    elif op == "lte" and actual_count > value:
                        return False
                    elif op == "eq" and actual_count != value:
                        return False
            return True
    
  3. Rules Engine:
    class RulesEngine:
        def __init__(self):
            self.rules: List[Rule] = []
            self.evaluator = ConditionEvaluator()
    
        def add_rule(self, **kwargs):
            rule = Rule(**kwargs)
            self.rules.append(rule)
            # Keep sorted by priority
            self.rules.sort(key=lambda r: r.priority, reverse=True)
    
        def validate(self, configuration: Dict) -> Dict:
            context = self._build_context(configuration)
            results = {
                "valid": True,
                "errors": [],
                "warnings": [],
                "suggestions": [],
                "auto_additions": [],
                "auto_removals": []
            }
    
            # Evaluate all rules
            for rule in self.rules:
                if not rule.enabled:
                    continue
    
                if self.evaluator.evaluate(rule.condition, context):
                    result = self._apply_rule(rule, context, results)
                    if result:
                        self._update_results(results, rule, result)
    
            results["valid"] = len(results["errors"]) == 0
            return results
    
        def _apply_rule(self, rule: Rule, context: Dict, current_results: Dict):
            action = rule.action
    
            if "require" in action:
                # Check if required item is present
                required = action["require"]
                if not self._is_present(required, context):
                    return {
                        "type": "error" if rule.type == RuleType.VALIDATION else "auto_add",
                        "missing": required,
                        "message": rule.message
                    }
    
            if "exclude" in action:
                excluded = action["exclude"]
                if self._is_present(excluded, context):
                    return {
                        "type": "error",
                        "conflict": excluded,
                        "message": rule.message
                    }
    
            if "suggest" in action:
                return {
                    "type": "suggestion",
                    "item": action["suggest"],
                    "message": rule.message
                }
    
            return None
    
  4. Dependency Graph (for complex dependencies):
    class DependencyGraph:
        def __init__(self):
            self.graph: Dict[str, Set[str]] = {}  # item -> items it requires
    
        def add_dependency(self, item: str, requires: str):
            if item not in self.graph:
                self.graph[item] = set()
            self.graph[item].add(requires)
    
        def get_all_dependencies(self, item: str) -> Set[str]:
            """Get transitive closure of all dependencies."""
            visited = set()
            to_visit = [item]
    
            while to_visit:
                current = to_visit.pop()
                if current in visited:
                    continue
                visited.add(current)
                for dep in self.graph.get(current, []):
                    if dep not in visited:
                        to_visit.append(dep)
    
            visited.remove(item)  # Don't include the item itself
            return visited
    
        def detect_cycles(self) -> List[List[str]]:
            """Detect circular dependencies."""
            cycles = []
            visited = set()
            rec_stack = set()
    
            def dfs(node, path):
                visited.add(node)
                rec_stack.add(node)
                path.append(node)
    
                for neighbor in self.graph.get(node, []):
                    if neighbor not in visited:
                        dfs(neighbor, path)
                    elif neighbor in rec_stack:
                        # Found cycle
                        cycle_start = path.index(neighbor)
                        cycles.append(path[cycle_start:] + [neighbor])
    
                path.pop()
                rec_stack.remove(node)
    
            for node in self.graph:
                if node not in visited:
                    dfs(node, [])
    
            return cycles
    

Learning milestones:

  1. Simple rules evaluate correctly → You understand condition matching
  2. Dependencies auto-add products → You understand inclusion rules
  3. Exclusions prevent conflicts → You understand constraint checking
  4. Suggestions work → You understand recommendation rules
  5. Performance with 100+ rules → You understand optimization

Project 3: Pricing Engine

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: Python
  • Alternative Programming Languages: TypeScript, Java, Go
  • Coolness Level: Level 3: Genuinely Clever
  • Business Potential: 4. The “Open Core” Infrastructure
  • Difficulty: Level 3: Advanced
  • Knowledge Area: Pricing Algorithms / Business Logic
  • Software or Tool: PostgreSQL, calculation engine
  • Main Book: “The Strategy and Tactics of Pricing” by Nagle & Holden

What you’ll build: A pricing engine that calculates prices from base rates, applies discounts (volume, customer, promotional), handles tiered pricing, bundle pricing, and determines when approvals are needed.

Why it teaches pricing complexity: Pricing in B2B is never simple. Multiple discount types, stacking rules, approval thresholds, margin calculations, and price list management are all critical business logic.

Core challenges you’ll face:

  • Discount stacking → maps to order of operations, compounding vs additive
  • Tiered vs volume pricing → maps to threshold calculations
  • Price list management → maps to customer segments, effective dates
  • Margin protection → maps to floor prices, approval triggers

Key Concepts:

  • Pricing Models: Tiered vs Volume Pricing
  • Discount Strategies: “The Strategy and Tactics of Pricing” - Nagle & Holden
  • Revenue Recognition: ASC 606 guidelines for software

Difficulty: Advanced Time estimate: 2-3 weeks Prerequisites: Project 1 completed, understanding of business math

Real world outcome:

# Configure pricing engine
pricing_engine = PricingEngine()

# Add price lists
pricing_engine.add_price_list(
    name="Standard 2024",
    effective_date="2024-01-01",
    products={
        "enterprise-license": {"base": 10000, "per_user": 100},
        "analytics-module": {"base": 3000},
        "api-module": {"base": 5000}
    }
)

# Add discount rules
pricing_engine.add_discount(
    name="Volume Discount",
    type="volume",
    tiers=[
        {"min": 0, "max": 25000, "discount": 0},
        {"min": 25001, "max": 50000, "discount": 0.05},
        {"min": 50001, "max": 100000, "discount": 0.10},
        {"min": 100001, "max": None, "discount": 0.15}
    ]
)

pricing_engine.add_discount(
    name="Gold Partner",
    type="customer_segment",
    segment="gold_partner",
    discount=0.05
)

# Calculate price
quote_items = [
    {"product": "enterprise-license", "quantity": 1, "users": 200},
    {"product": "analytics-module", "quantity": 1},
    {"product": "api-module", "quantity": 1}
]

result = pricing_engine.calculate(
    items=quote_items,
    customer_segment="gold_partner",
    promo_code="Q4SPECIAL"
)

# Result:
# {
#     "line_items": [
#         {"product": "enterprise-license", "list_price": 30000, "net_price": 24300},
#         {"product": "analytics-module", "list_price": 3000, "net_price": 2430},
#         {"product": "api-module", "list_price": 5000, "net_price": 4050}
#     ],
#     "subtotal": 38000,
#     "discounts": [
#         {"name": "Volume Discount", "type": "volume", "amount": 3800, "percent": 10},
#         {"name": "Gold Partner", "type": "customer", "amount": 1710, "percent": 5},
#         {"name": "Q4 Special", "type": "promo", "amount": 1500, "fixed": True}
#     ],
#     "total_discount": 7010,
#     "net_total": 30990,
#     "margin": 18594,
#     "margin_percent": 60,
#     "requires_approval": False,
#     "approval_reason": None
# }

Implementation Hints:

  1. Price List Data Model:
    @dataclass
    class PriceListEntry:
        product_id: str
        base_price: Decimal
        unit_price: Optional[Decimal] = None  # Per-unit pricing
        price_type: str = "fixed"  # fixed, per_unit, tiered
        tiers: Optional[List[Dict]] = None
        effective_from: date = None
        effective_to: date = None
        currency: str = "USD"
    
    @dataclass
    class PriceList:
        id: str
        name: str
        entries: Dict[str, PriceListEntry]
        segment: Optional[str] = None  # Customer segment this applies to
        priority: int = 100
    
    @dataclass
    class Discount:
        id: str
        name: str
        type: str  # volume, customer, promo, bundle, manual
        value: Decimal  # Percent or fixed amount
        is_percent: bool = True
        conditions: Dict = None
        stackable: bool = True
        priority: int = 100  # Order of application
        min_margin: Optional[Decimal] = None  # Floor
    
  2. Pricing Calculation:
    class PricingEngine:
        def calculate(self, items: List[Dict], **context) -> Dict:
            result = {
                "line_items": [],
                "subtotal": Decimal("0"),
                "discounts": [],
                "net_total": Decimal("0"),
                "requires_approval": False
            }
    
            # Step 1: Calculate list prices
            for item in items:
                line = self._calculate_line_item(item, context)
                result["line_items"].append(line)
                result["subtotal"] += line["list_price"]
    
            # Step 2: Apply discounts (in priority order)
            applicable_discounts = self._get_applicable_discounts(result, context)
            running_total = result["subtotal"]
    
            for discount in sorted(applicable_discounts, key=lambda d: d.priority):
                discount_amount = self._calculate_discount(
                    discount, running_total, result, context
                )
                if discount_amount > 0:
                    result["discounts"].append({
                        "name": discount.name,
                        "type": discount.type,
                        "amount": discount_amount,
                        "percent": (discount_amount / result["subtotal"] * 100)
                    })
                    if discount.stackable:
                        running_total -= discount_amount
    
            result["total_discount"] = result["subtotal"] - running_total
            result["net_total"] = running_total
    
            # Step 3: Calculate margin and check approval
            result["margin"] = self._calculate_margin(result)
            result["margin_percent"] = (result["margin"] / result["net_total"] * 100)
            result["requires_approval"] = self._check_approval_needed(result, context)
    
            return result
    
        def _calculate_line_item(self, item: Dict, context: Dict) -> Dict:
            product = item["product"]
            price_entry = self._get_price(product, context)
    
            if price_entry.price_type == "fixed":
                list_price = price_entry.base_price * item.get("quantity", 1)
            elif price_entry.price_type == "per_unit":
                units = item.get("users") or item.get("units") or 1
                list_price = price_entry.base_price + (price_entry.unit_price * units)
            elif price_entry.price_type == "tiered":
                list_price = self._calculate_tiered_price(price_entry, item)
    
            return {
                "product": product,
                "quantity": item.get("quantity", 1),
                "list_price": list_price
            }
    
  3. Tiered vs Volume Pricing:
    def _calculate_tiered_price(self, entry: PriceListEntry, item: Dict) -> Decimal:
        """Tiered: each tier has its own price, applied progressively."""
        quantity = item.get("quantity", 1)
        total = Decimal("0")
        remaining = quantity
    
        for tier in sorted(entry.tiers, key=lambda t: t["min"]):
            tier_min = tier["min"]
            tier_max = tier.get("max") or float("inf")
            tier_price = Decimal(str(tier["price"]))
    
            if remaining <= 0:
                break
    
            tier_quantity = min(remaining, tier_max - tier_min + 1)
            total += tier_quantity * tier_price
            remaining -= tier_quantity
    
        return total
    
    def _calculate_volume_price(self, entry: PriceListEntry, item: Dict) -> Decimal:
        """Volume: all units at the price of the tier you land in."""
        quantity = item.get("quantity", 1)
    
        for tier in sorted(entry.tiers, key=lambda t: t["min"], reverse=True):
            if quantity >= tier["min"]:
                return quantity * Decimal(str(tier["price"]))
    
        return quantity * entry.base_price
    
  4. Discount Stacking Logic:
    def _apply_discounts(self, discounts: List[Discount], subtotal: Decimal) -> Decimal:
        """Apply discounts with proper stacking rules."""
        # Group by stackability
        stackable = [d for d in discounts if d.stackable]
        non_stackable = [d for d in discounts if not d.stackable]
    
        # For non-stackable, take the best one
        if non_stackable:
            best_non_stackable = max(non_stackable, key=lambda d: self._discount_value(d, subtotal))
            non_stackable = [best_non_stackable]
    
        # Apply stackable discounts in order (compounding)
        current = subtotal
        for discount in sorted(stackable, key=lambda d: d.priority):
            if discount.is_percent:
                current = current * (1 - discount.value / 100)
            else:
                current = current - discount.value
    
        # Apply best non-stackable
        for discount in non_stackable:
            if discount.is_percent:
                current = current * (1 - discount.value / 100)
            else:
                current = current - discount.value
    
        return subtotal - current  # Total discount amount
    
  5. Approval Rules:
    def _check_approval_needed(self, result: Dict, context: Dict) -> bool:
        discount_percent = (result["total_discount"] / result["subtotal"]) * 100
    
        # Check discount thresholds
        if discount_percent > 25:
            result["approval_reason"] = f"Discount {discount_percent:.1f}% exceeds 25% (requires VP)"
            result["approval_level"] = "vp_sales"
            return True
        elif discount_percent > 15:
            result["approval_reason"] = f"Discount {discount_percent:.1f}% exceeds 15%"
            result["approval_level"] = "sales_manager"
            return True
    
        # Check margin floor
        if result["margin_percent"] < 40:
            result["approval_reason"] = f"Margin {result['margin_percent']:.1f}% below 40%"
            result["approval_level"] = "finance"
            return True
    
        return False
    

Learning milestones:

  1. Base pricing works → You understand price lists
  2. Volume/tiered pricing works → You understand complex pricing
  3. Discounts stack correctly → You understand discount logic
  4. Approval rules trigger → You understand business controls
  5. Margin calculations correct → You understand profitability

Project 4: Quote Builder & Document Generator

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: Python (with WeasyPrint/ReportLab)
  • Alternative Programming Languages: TypeScript (PDFKit), Java (iText), C# (IronPDF)
  • Coolness Level: Level 3: Genuinely Clever
  • Business Potential: 3. The “Service & Support” Model
  • Difficulty: Level 2: Intermediate
  • Knowledge Area: Document Generation / Templating
  • Software or Tool: WeasyPrint, Jinja2, HTML/CSS
  • Main Book: None (practical implementation)

What you’ll build: A quote builder that lets sales reps create quotes from configurations, and generates professional PDF documents with branding, terms, and e-signature placeholders.

Why it teaches document generation: The quote is the customer-facing output of CPQ. Professional, accurate, branded documents are essential for sales success and legal compliance.

Core challenges you’ll face:

  • Template design → maps to HTML/CSS to PDF, page breaks
  • Dynamic content → maps to conditional sections, loops
  • Branding → maps to logos, fonts, colors
  • Multiple formats → maps to PDF, Word, HTML

Key Concepts:

  • PDF Generation: WeasyPrint / ReportLab documentation
  • Templating: Jinja2 documentation
  • Document Design: “The Non-Designer’s Design Book” - Robin Williams

Difficulty: Intermediate Time estimate: 1-2 weeks Prerequisites: Project 3 completed, HTML/CSS knowledge

Real world outcome:

# Create a quote
quote = quote_builder.create_quote(
    customer={
        "name": "Acme Corporation",
        "contact": "John Smith",
        "email": "john@acme.com",
        "address": "123 Business Ave, New York, NY 10001"
    },
    sales_rep={
        "name": "Jane Doe",
        "email": "jane@ourcompany.com",
        "phone": "555-123-4567"
    },
    configuration=validated_config,
    pricing=pricing_result,
    valid_days=30,
    terms="standard"
)

# Generate PDF
pdf_bytes = quote_generator.generate_pdf(
    quote,
    template="professional",
    options={
        "include_terms": True,
        "include_signature_block": True,
        "watermark": None  # or "DRAFT"
    }
)

# Save or send
with open(f"Quote_{quote.number}.pdf", "wb") as f:
    f.write(pdf_bytes)

# Or email directly
email_service.send_quote(quote, pdf_bytes)

Implementation Hints:

  1. Quote Data Model:
    @dataclass
    class Quote:
        id: UUID
        number: str  # Q-2024-00001
        version: int = 1
        status: str = "draft"  # draft, pending_approval, approved, sent, accepted, rejected
    
        customer: Customer
        sales_rep: User
    
        line_items: List[QuoteLineItem]
        subtotal: Decimal
        discounts: List[AppliedDiscount]
        total: Decimal
    
        valid_from: date
        valid_until: date
        terms_id: str
    
        created_at: datetime
        updated_at: datetime
        sent_at: Optional[datetime] = None
        accepted_at: Optional[datetime] = None
    
    @dataclass
    class QuoteLineItem:
        product_id: str
        product_name: str
        description: str
        quantity: int
        unit_price: Decimal
        total_price: Decimal
        configuration: Dict  # Selected options
    
  2. Quote Template (Jinja2 + HTML):
    <!DOCTYPE html>
    <html>
    <head>
        <style>
            @page {
                size: letter;
                margin: 1in;
                @bottom-center {
                    content: "Page " counter(page) " of " counter(pages);
                }
            }
            body { font-family: 'Helvetica', sans-serif; }
            .header { display: flex; justify-content: space-between; }
            .logo { height: 60px; }
            .quote-info { text-align: right; }
            .parties { display: flex; justify-content: space-between; margin: 2em 0; }
            table { width: 100%; border-collapse: collapse; margin: 2em 0; }
            th, td { border: 1px solid #ddd; padding: 0.5em; text-align: left; }
            th { background: #f5f5f5; }
            .totals { text-align: right; }
            .signature-block { margin-top: 3em; page-break-inside: avoid; }
        </style>
    </head>
    <body>
        <div class="header">
            <img src="{{ company.logo_url }}" class="logo" alt="{{ company.name }}">
            <div class="quote-info">
                <h1>Quote</h1>
                <p><strong>Quote #:</strong> {{ quote.number }}</p>
                <p><strong>Date:</strong> {{ quote.created_at | date }}</p>
                <p><strong>Valid Until:</strong> {{ quote.valid_until | date }}</p>
            </div>
        </div>
    
        <div class="parties">
            <div class="customer">
                <h3>Prepared For:</h3>
                <p><strong>{{ quote.customer.name }}</strong></p>
                <p>{{ quote.customer.contact }}</p>
                <p>{{ quote.customer.address }}</p>
            </div>
            <div class="sales-rep">
                <h3>Prepared By:</h3>
                <p><strong>{{ quote.sales_rep.name }}</strong></p>
                <p>{{ quote.sales_rep.email }}</p>
                <p>{{ quote.sales_rep.phone }}</p>
            </div>
        </div>
    
        <table>
            <thead>
                <tr>
                    <th>Product</th>
                    <th>Description</th>
                    <th>Qty</th>
                    <th>Unit Price</th>
                    <th>Total</th>
                </tr>
            </thead>
            <tbody>
                {% for item in quote.line_items %}
                <tr>
                    <td>{{ item.product_name }}</td>
                    <td>{{ item.description }}</td>
                    <td>{{ item.quantity }}</td>
                    <td>{{ item.unit_price | currency }}</td>
                    <td>{{ item.total_price | currency }}</td>
                </tr>
                {% endfor %}
            </tbody>
            <tfoot>
                <tr>
                    <td colspan="4" class="totals"><strong>Subtotal:</strong></td>
                    <td>{{ quote.subtotal | currency }}</td>
                </tr>
                {% for discount in quote.discounts %}
                <tr>
                    <td colspan="4" class="totals">{{ discount.name }}:</td>
                    <td>-{{ discount.amount | currency }}</td>
                </tr>
                {% endfor %}
                <tr>
                    <td colspan="4" class="totals"><strong>Total:</strong></td>
                    <td><strong>{{ quote.total | currency }}</strong></td>
                </tr>
            </tfoot>
        </table>
    
        {% if include_terms %}
        <div class="terms">
            <h3>Terms & Conditions</h3>
            {{ terms_content | safe }}
        </div>
        {% endif %}
    
        {% if include_signature_block %}
        <div class="signature-block">
            <h3>Acceptance</h3>
            <p>By signing below, you agree to the terms of this quote.</p>
            <div style="margin-top: 2em;">
                <p>___________________________ &nbsp;&nbsp;&nbsp; _______________</p>
                <p>Signature &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Date</p>
                <p style="margin-top: 1em;">___________________________</p>
                <p>Printed Name</p>
            </div>
        </div>
        {% endif %}
    </body>
    </html>
    
  3. PDF Generator:
    from weasyprint import HTML, CSS
    from jinja2 import Environment, FileSystemLoader
    
    class QuoteGenerator:
        def __init__(self, templates_dir: str):
            self.env = Environment(loader=FileSystemLoader(templates_dir))
            self.env.filters['currency'] = lambda x: f"${x:,.2f}"
            self.env.filters['date'] = lambda x: x.strftime("%B %d, %Y")
    
        def generate_pdf(self, quote: Quote, template: str = "default", **options) -> bytes:
            # Load template
            template = self.env.get_template(f"{template}.html")
    
            # Render HTML
            html_content = template.render(
                quote=quote,
                company=self._get_company_info(),
                terms_content=self._get_terms(quote.terms_id),
                **options
            )
    
            # Convert to PDF
            pdf = HTML(string=html_content).write_pdf(
                stylesheets=[CSS(filename='styles/quote.css')]
            )
    
            return pdf
    
        def generate_word(self, quote: Quote, template: str = "default") -> bytes:
            """Generate Word document using python-docx."""
            from docx import Document
            from docx.shared import Inches, Pt
    
            doc = Document()
            # ... build Word document
            return doc
    
        def _add_watermark(self, pdf_bytes: bytes, text: str) -> bytes:
            """Add watermark like 'DRAFT' to PDF."""
            from PyPDF2 import PdfReader, PdfWriter
            # ... add watermark
            return pdf_bytes
    
  4. Quote Versioning:
    class QuoteService:
        def create_version(self, quote: Quote) -> Quote:
            """Create a new version of a quote."""
            new_quote = Quote(
                id=uuid4(),
                number=quote.number,
                version=quote.version + 1,
                parent_id=quote.id,
                # Copy other fields...
            )
            return new_quote
    
        def get_version_history(self, quote_number: str) -> List[Quote]:
            return self.db.query(Quote).filter(
                Quote.number == quote_number
            ).order_by(Quote.version).all()
    

Learning milestones:

  1. Basic PDF generates → You understand PDF generation
  2. Template renders correctly → You understand templating
  3. Multiple templates work → You understand template management
  4. Versioning works → You understand quote lifecycle
  5. Professional output → Ready for production use

Project 5: Approval Workflow Engine

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: Python
  • Alternative Programming Languages: TypeScript, Java, Go
  • Coolness Level: Level 3: Genuinely Clever
  • Business Potential: 3. The “Service & Support” Model
  • Difficulty: Level 3: Advanced
  • Knowledge Area: Workflow Engines / State Machines
  • Software or Tool: Custom or Temporal/Camunda
  • Main Book: “Enterprise Integration Patterns” by Hohpe & Woolf

What you’ll build: An approval workflow engine that routes quotes for approval based on rules, handles sequential and parallel approvals, escalations, and delegations.

Why it teaches workflow: Approval workflows are critical business controls. Understanding state machines, routing rules, and notification systems is essential for enterprise software.

Core challenges you’ll face:

  • State machine design → maps to quote/approval states, transitions
  • Routing rules → maps to determining who approves what
  • Parallel vs sequential → maps to AND vs OR approval logic
  • Escalation & delegation → maps to timeout handling, out-of-office

Key Concepts:

Difficulty: Advanced Time estimate: 2-3 weeks Prerequisites: Project 3 completed, understanding of state machines

Real world outcome:

# Define approval workflow
workflow_engine.create_workflow(
    name="Quote Approval",
    trigger={"entity": "quote", "event": "submitted"},
    steps=[
        {
            "name": "Manager Approval",
            "type": "approval",
            "assignee": {"type": "role", "role": "sales_manager", "scope": "submitter_manager"},
            "conditions": [
                {"field": "discount_percent", "operator": "gt", "value": 10}
            ],
            "timeout": {"hours": 24, "action": "escalate"},
            "on_approve": "next",
            "on_reject": "end_rejected"
        },
        {
            "name": "Finance Review",
            "type": "approval",
            "assignee": {"type": "role", "role": "finance"},
            "conditions": [
                {"field": "margin_percent", "operator": "lt", "value": 40}
            ],
            "timeout": {"hours": 48, "action": "escalate"},
            "on_approve": "next",
            "on_reject": "end_rejected"
        },
        {
            "name": "VP Approval",
            "type": "approval",
            "assignee": {"type": "role", "role": "vp_sales"},
            "conditions": [
                {"field": "total", "operator": "gt", "value": 100000}
            ],
            "timeout": {"hours": 72, "action": "notify"},
            "on_approve": "end_approved",
            "on_reject": "end_rejected"
        }
    ]
)

# Submit quote for approval
result = workflow_engine.submit(quote)
# {
#     "workflow_id": "wf_123",
#     "status": "pending_approval",
#     "current_step": "Manager Approval",
#     "assignee": "John Manager",
#     "pending_since": "2024-12-21T10:00:00Z"
# }

# Approve
workflow_engine.approve(workflow_id="wf_123", user=manager, comment="Looks good")
# Moves to next step or completes

Implementation Hints:

  1. Workflow Data Model:
    class WorkflowStatus(Enum):
        PENDING = "pending"
        IN_PROGRESS = "in_progress"
        APPROVED = "approved"
        REJECTED = "rejected"
        CANCELLED = "cancelled"
        ESCALATED = "escalated"
    
    @dataclass
    class WorkflowDefinition:
        id: str
        name: str
        entity_type: str  # quote, order, contract
        trigger: Dict
        steps: List[WorkflowStep]
        enabled: bool = True
    
    @dataclass
    class WorkflowStep:
        name: str
        type: str  # approval, notification, action
        assignee: Dict  # Who handles this step
        conditions: List[Dict]  # When this step applies
        timeout: Optional[Dict]
        actions: Dict  # What happens on approve/reject
    
    @dataclass
    class WorkflowInstance:
        id: UUID
        definition_id: str
        entity_type: str
        entity_id: UUID
        status: WorkflowStatus
        current_step: int
        history: List[WorkflowEvent]
        created_at: datetime
        updated_at: datetime
    
    @dataclass
    class ApprovalRequest:
        id: UUID
        workflow_id: UUID
        step_name: str
        assignee_id: UUID
        status: str  # pending, approved, rejected
        comment: Optional[str]
        created_at: datetime
        responded_at: Optional[datetime]
    
  2. Workflow Engine:
    class WorkflowEngine:
        def submit(self, entity_type: str, entity_id: UUID, entity_data: Dict) -> WorkflowInstance:
            # Find matching workflow definition
            definition = self._find_workflow(entity_type, entity_data)
            if not definition:
                return None  # No workflow needed
    
            # Create instance
            instance = WorkflowInstance(
                id=uuid4(),
                definition_id=definition.id,
                entity_type=entity_type,
                entity_id=entity_id,
                status=WorkflowStatus.IN_PROGRESS,
                current_step=0,
                history=[]
            )
    
            # Start first applicable step
            self._advance_to_next_step(instance, entity_data)
    
            self.db.add(instance)
            self.db.commit()
    
            return instance
    
        def _advance_to_next_step(self, instance: WorkflowInstance, entity_data: Dict):
            definition = self._get_definition(instance.definition_id)
    
            while instance.current_step < len(definition.steps):
                step = definition.steps[instance.current_step]
    
                # Check if step applies
                if self._evaluate_conditions(step.conditions, entity_data):
                    # Create approval request
                    assignee = self._resolve_assignee(step.assignee, entity_data)
                    request = ApprovalRequest(
                        workflow_id=instance.id,
                        step_name=step.name,
                        assignee_id=assignee.id
                    )
                    self.db.add(request)
    
                    # Send notification
                    self._notify_approver(assignee, instance, step)
    
                    # Schedule timeout if configured
                    if step.timeout:
                        self._schedule_timeout(instance.id, step.timeout)
    
                    return  # Wait for approval
    
                # Step doesn't apply, skip to next
                instance.current_step += 1
    
            # No more steps, workflow complete
            instance.status = WorkflowStatus.APPROVED
            self._on_workflow_complete(instance)
    
        def approve(self, workflow_id: UUID, user: User, comment: str = None):
            instance = self._get_instance(workflow_id)
            request = self._get_pending_request(workflow_id)
    
            # Verify user can approve
            if not self._can_approve(request, user):
                raise PermissionError("User cannot approve this request")
    
            # Record approval
            request.status = "approved"
            request.responded_at = datetime.now()
            request.comment = comment
    
            instance.history.append(WorkflowEvent(
                type="approved",
                step=request.step_name,
                user_id=user.id,
                comment=comment,
                timestamp=datetime.now()
            ))
    
            # Advance to next step
            instance.current_step += 1
            entity_data = self._get_entity_data(instance)
            self._advance_to_next_step(instance, entity_data)
    
            self.db.commit()
    
        def reject(self, workflow_id: UUID, user: User, reason: str):
            instance = self._get_instance(workflow_id)
            request = self._get_pending_request(workflow_id)
    
            request.status = "rejected"
            request.responded_at = datetime.now()
            request.comment = reason
    
            instance.status = WorkflowStatus.REJECTED
            instance.history.append(WorkflowEvent(
                type="rejected",
                step=request.step_name,
                user_id=user.id,
                comment=reason
            ))
    
            self._on_workflow_rejected(instance, reason)
            self.db.commit()
    
  3. Assignee Resolution:
    def _resolve_assignee(self, assignee_config: Dict, entity_data: Dict) -> User:
        assignee_type = assignee_config["type"]
    
        if assignee_type == "user":
            return self._get_user(assignee_config["user_id"])
    
        elif assignee_type == "role":
            role = assignee_config["role"]
            scope = assignee_config.get("scope")
    
            if scope == "submitter_manager":
                submitter = entity_data.get("submitter")
                return self._get_manager(submitter)
            elif scope == "region":
                region = entity_data.get("region")
                return self._get_role_in_region(role, region)
            else:
                return self._get_any_with_role(role)
    
        elif assignee_type == "dynamic":
            # Evaluate expression to determine assignee
            expr = assignee_config["expression"]
            return self._evaluate_assignee_expression(expr, entity_data)
    
  4. Timeout & Escalation:
    def _schedule_timeout(self, workflow_id: UUID, timeout_config: Dict):
        delay = timedelta(hours=timeout_config.get("hours", 24))
        action = timeout_config.get("action", "notify")
    
        self.scheduler.schedule(
            task=self._handle_timeout,
            args=(workflow_id, action),
            run_at=datetime.now() + delay
        )
    
    def _handle_timeout(self, workflow_id: UUID, action: str):
        instance = self._get_instance(workflow_id)
        request = self._get_pending_request(workflow_id)
    
        if request.status != "pending":
            return  # Already handled
    
        if action == "escalate":
            # Reassign to escalation target
            new_assignee = self._get_escalation_target(request)
            request.assignee_id = new_assignee.id
            self._notify_approver(new_assignee, instance, is_escalation=True)
    
        elif action == "auto_approve":
            self.approve(workflow_id, system_user, "Auto-approved after timeout")
    
        elif action == "notify":
            self._send_reminder(request)
            # Reschedule another reminder
            self._schedule_timeout(workflow_id, {"hours": 24, "action": "escalate"})
    

Learning milestones:

  1. Basic approval flow works → You understand workflows
  2. Multiple steps work → You understand routing
  3. Conditions skip steps correctly → You understand rules
  4. Escalation works → You understand timeouts
  5. Parallel approvals work → You understand complex workflows

Project 6: Guided Selling Assistant

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: TypeScript (React)
  • Alternative Programming Languages: Python (Streamlit), Vue.js
  • Coolness Level: Level 3: Genuinely Clever
  • Business Potential: 3. The “Service & Support” Model
  • Difficulty: Level 2: Intermediate
  • Knowledge Area: UX Design / Decision Trees
  • Software or Tool: React, decision tree engine
  • Main Book: “Don’t Make Me Think” by Steve Krug

What you’ll build: A guided selling interface that helps sales reps configure products by asking questions, making recommendations, and preventing mistakes—like a “wizard” for complex configurations.

Why it teaches sales UX: CPQ is only useful if sales reps can use it easily. Guided selling transforms complex configuration into simple Q&A, reducing training and errors.

Core challenges you’ll face:

  • Question flow design → maps to decision trees, branching logic
  • Recommendation engine → maps to rules, ML-based suggestions
  • Progress tracking → maps to multi-step wizards, state management
  • Mobile-friendly design → maps to responsive UI, touch interactions

Key Concepts:

  • Decision Trees: Basic AI/ML concepts
  • Wizard UX Patterns: “Designing Interfaces” - Jenifer Tidwell
  • Sales Psychology: “SPIN Selling” - Neil Rackham

Difficulty: Intermediate Time estimate: 2 weeks Prerequisites: React knowledge, Projects 1-2 completed

Real world outcome:

// Guided selling configuration
const sellingGuide = {
  name: "Enterprise Software Selection",
  steps: [
    {
      id: "company-size",
      question: "How many employees does your customer have?",
      type: "single-select",
      options: [
        { value: "small", label: "1-50 employees", icon: "👥" },
        { value: "medium", label: "51-500 employees", icon: "🏢" },
        { value: "large", label: "500+ employees", icon: "🏙️" }
      ],
      recommendation: {
        condition: { answer: "large" },
        message: "Enterprise customers typically benefit from our Premium support package."
      }
    },
    {
      id: "primary-need",
      question: "What's the customer's primary need?",
      type: "multi-select",
      options: [
        { value: "analytics", label: "Data Analytics" },
        { value: "automation", label: "Process Automation" },
        { value: "collaboration", label: "Team Collaboration" },
        { value: "security", label: "Security & Compliance" }
      ]
    },
    {
      id: "budget",
      question: "What's the expected budget range?",
      type: "single-select",
      showIf: { "company-size": ["medium", "large"] },
      options: [
        { value: "tier1", label: "$10K - $25K/year" },
        { value: "tier2", label: "$25K - $75K/year" },
        { value: "tier3", label: "$75K+/year" }
      ]
    }
  ],
  completion: {
    action: "recommend-configuration",
    rules: [
      {
        conditions: { "company-size": "large", "budget": "tier3" },
        configuration: "enterprise-premium"
      },
      // ... more rules
    ]
  }
};

// React component usage
<GuidedSelling
  guide={sellingGuide}
  onComplete={(answers, recommendation) => {
    setConfiguration(recommendation);
    navigateToQuoteBuilder();
  }}
/>

Implementation Hints:

  1. React Component Structure:
    interface GuidedSellingProps {
      guide: SellingGuide;
      onComplete: (answers: Answers, recommendation: Configuration) => void;
    }
    
    const GuidedSelling: React.FC<GuidedSellingProps> = ({ guide, onComplete }) => {
      const [currentStep, setCurrentStep] = useState(0);
      const [answers, setAnswers] = useState<Answers>({});
      const [recommendations, setRecommendations] = useState<Recommendation[]>([]);
    
      const visibleSteps = useMemo(() =>
        guide.steps.filter(step => evaluateShowIf(step.showIf, answers)),
        [guide.steps, answers]
      );
    
      const handleAnswer = (stepId: string, value: any) => {
        setAnswers(prev => ({ ...prev, [stepId]: value }));
    
        // Check for recommendations
        const step = guide.steps.find(s => s.id === stepId);
        if (step?.recommendation) {
          const rec = evaluateRecommendation(step.recommendation, value);
          if (rec) {
            setRecommendations(prev => [...prev, rec]);
          }
        }
    
        // Move to next step
        setCurrentStep(prev => prev + 1);
      };
    
      const currentStepData = visibleSteps[currentStep];
    
      if (!currentStepData) {
        // All steps complete
        const recommendation = calculateRecommendation(guide.completion, answers);
        return (
          <CompletionScreen
            answers={answers}
            recommendation={recommendation}
            onConfirm={() => onComplete(answers, recommendation)}
          />
        );
      }
    
      return (
        <div className="guided-selling">
          <ProgressBar current={currentStep} total={visibleSteps.length} />
          <QuestionCard step={currentStepData} onAnswer={handleAnswer} />
          {recommendations.length > 0 && (
            <RecommendationPanel recommendations={recommendations} />
          )}
        </div>
      );
    };
    
  2. Question Components:
    const QuestionCard: React.FC<{ step: Step; onAnswer: (id: string, value: any) => void }> = ({
      step,
      onAnswer
    }) => {
      const [selected, setSelected] = useState<any>(null);
    
      return (
        <div className="question-card">
          <h2>{step.question}</h2>
    
          {step.type === "single-select" && (
            <div className="options-grid">
              {step.options.map(opt => (
                <OptionButton
                  key={opt.value}
                  option={opt}
                  selected={selected === opt.value}
                  onClick={() => {
                    setSelected(opt.value);
                    onAnswer(step.id, opt.value);
                  }}
                />
              ))}
            </div>
          )}
    
          {step.type === "multi-select" && (
            <div className="options-grid">
              {step.options.map(opt => (
                <CheckboxOption
                  key={opt.value}
                  option={opt}
                  checked={selected?.includes(opt.value)}
                  onChange={(checked) => {
                    const newValue = checked
                      ? [...(selected || []), opt.value]
                      : (selected || []).filter(v => v !== opt.value);
                    setSelected(newValue);
                  }}
                />
              ))}
              <button onClick={() => onAnswer(step.id, selected)}>Continue</button>
            </div>
          )}
        </div>
      );
    };
    
  3. Recommendation Engine:
    function calculateRecommendation(completion: Completion, answers: Answers): Configuration {
      // Try each rule in order
      for (const rule of completion.rules) {
        if (evaluateConditions(rule.conditions, answers)) {
          return getConfiguration(rule.configuration);
        }
      }
    
      // Default fallback
      return getDefaultConfiguration();
    }
    
    function evaluateConditions(conditions: Conditions, answers: Answers): boolean {
      for (const [key, expected] of Object.entries(conditions)) {
        const actual = answers[key];
    
        if (Array.isArray(expected)) {
          if (!expected.includes(actual)) return false;
        } else {
          if (actual !== expected) return false;
        }
      }
      return true;
    }
    

Learning milestones:

  1. Basic wizard flow works → You understand step navigation
  2. Conditional steps work → You understand branching
  3. Recommendations appear → You understand decision logic
  4. Final configuration correct → You understand the full flow
  5. Mobile responsive → Ready for production

Project 7: CRM Integration (Salesforce)

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: Python + Apex (Salesforce)
  • Alternative Programming Languages: TypeScript, Java
  • Coolness Level: Level 3: Genuinely Clever
  • Business Potential: 4. The “Open Core” Infrastructure
  • Difficulty: Level 3: Advanced
  • Knowledge Area: Integration / APIs
  • Software or Tool: Salesforce, REST APIs, OAuth
  • Main Book: “Enterprise Integration Patterns” by Hohpe & Woolf

What you’ll build: Integration between your CPQ system and Salesforce CRM—syncing opportunities, contacts, and pushing quotes back to Salesforce as Quote objects.

Why it teaches integration: Real CPQ systems don’t exist in isolation. CRM integration is essential for sales workflows, and understanding OAuth, webhooks, and data sync patterns is critical for enterprise software.

Core challenges you’ll face:

  • OAuth authentication → maps to Salesforce Connected Apps
  • Data mapping → maps to field mapping, transformations
  • Bidirectional sync → maps to conflict resolution, timestamps
  • Bulk operations → maps to Salesforce limits, batching

Key Concepts:

  • Salesforce API: Salesforce REST API Guide
  • OAuth 2.0: RFC 6749
  • Integration Patterns: “Enterprise Integration Patterns” - Hohpe & Woolf

Difficulty: Advanced Time estimate: 2-3 weeks Prerequisites: Previous projects, understanding of REST APIs

Real world outcome:

# Connect to Salesforce
sf_client = SalesforceClient(
    client_id=os.environ["SF_CLIENT_ID"],
    client_secret=os.environ["SF_CLIENT_SECRET"],
    username=os.environ["SF_USERNAME"],
    password=os.environ["SF_PASSWORD"]
)

# Sync opportunity from Salesforce
opportunity = sf_client.get_opportunity("0061R00001234")
local_opportunity = cpq.sync_opportunity(opportunity)

# Create quote in CPQ
quote = cpq.create_quote(
    opportunity_id=local_opportunity.id,
    configuration=config,
    pricing=pricing_result
)

# Push quote back to Salesforce
sf_quote = sf_client.create_quote({
    "OpportunityId": opportunity["Id"],
    "Name": quote.number,
    "ExpirationDate": quote.valid_until.isoformat(),
    "Status": "Draft",
    "Description": "Created from CPQ System",
    "GrandTotal": float(quote.total)
})

# Attach PDF
sf_client.attach_file(
    parent_id=sf_quote["Id"],
    filename=f"Quote_{quote.number}.pdf",
    content=pdf_bytes,
    content_type="application/pdf"
)

Implementation Hints:

  1. Salesforce OAuth Client:
    import requests
    from urllib.parse import urlencode
    
    class SalesforceClient:
        def __init__(self, client_id, client_secret, username, password, sandbox=False):
            self.client_id = client_id
            self.client_secret = client_secret
            self.base_url = "https://test.salesforce.com" if sandbox else "https://login.salesforce.com"
            self.access_token = None
            self.instance_url = None
            self._authenticate(username, password)
    
        def _authenticate(self, username, password):
            response = requests.post(
                f"{self.base_url}/services/oauth2/token",
                data={
                    "grant_type": "password",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "username": username,
                    "password": password
                }
            )
            response.raise_for_status()
            data = response.json()
            self.access_token = data["access_token"]
            self.instance_url = data["instance_url"]
    
        def _request(self, method, path, **kwargs):
            headers = {
                "Authorization": f"Bearer {self.access_token}",
                "Content-Type": "application/json"
            }
            url = f"{self.instance_url}{path}"
            response = requests.request(method, url, headers=headers, **kwargs)
            response.raise_for_status()
            return response.json() if response.content else None
    
        def get_opportunity(self, opportunity_id):
            return self._request("GET", f"/services/data/v58.0/sobjects/Opportunity/{opportunity_id}")
    
        def create_quote(self, data):
            return self._request("POST", "/services/data/v58.0/sobjects/Quote", json=data)
    
        def attach_file(self, parent_id, filename, content, content_type):
            import base64
            attachment = {
                "ParentId": parent_id,
                "Name": filename,
                "Body": base64.b64encode(content).decode(),
                "ContentType": content_type
            }
            return self._request("POST", "/services/data/v58.0/sobjects/Attachment", json=attachment)
    
  2. Data Mapper:
    class DataMapper:
        def __init__(self, mappings_config):
            self.mappings = mappings_config
    
        def sf_to_local(self, sf_object, object_type):
            mapping = self.mappings[object_type]["sf_to_local"]
            result = {}
    
            for local_field, sf_config in mapping.items():
                if isinstance(sf_config, str):
                    # Simple field mapping
                    result[local_field] = sf_object.get(sf_config)
                elif isinstance(sf_config, dict):
                    # Complex transformation
                    value = sf_object.get(sf_config["field"])
                    if sf_config.get("transform"):
                        value = self._transform(value, sf_config["transform"])
                    result[local_field] = value
    
            return result
    
        def local_to_sf(self, local_object, object_type):
            mapping = self.mappings[object_type]["local_to_sf"]
            result = {}
    
            for sf_field, local_config in mapping.items():
                # Similar logic...
                pass
    
            return result
    
    # Example mappings config
    MAPPINGS = {
        "opportunity": {
            "sf_to_local": {
                "name": "Name",
                "amount": "Amount",
                "stage": "StageName",
                "close_date": {"field": "CloseDate", "transform": "parse_date"},
                "customer_id": "AccountId"
            },
            "local_to_sf": {
                "Name": "name",
                "Amount": "amount"
            }
        }
    }
    
  3. Sync Service:
    class SyncService:
        def __init__(self, sf_client, cpq_db, mapper):
            self.sf = sf_client
            self.db = cpq_db
            self.mapper = mapper
    
        def sync_opportunity(self, sf_opportunity_id):
            # Get from Salesforce
            sf_opp = self.sf.get_opportunity(sf_opportunity_id)
    
            # Map to local format
            local_data = self.mapper.sf_to_local(sf_opp, "opportunity")
    
            # Find or create local record
            existing = self.db.query(Opportunity).filter(
                Opportunity.sf_id == sf_opportunity_id
            ).first()
    
            if existing:
                # Update if SF is newer
                sf_modified = parse_datetime(sf_opp["SystemModstamp"])
                if sf_modified > existing.updated_at:
                    for key, value in local_data.items():
                        setattr(existing, key, value)
                    existing.updated_at = datetime.now()
            else:
                # Create new
                existing = Opportunity(sf_id=sf_opportunity_id, **local_data)
                self.db.add(existing)
    
            self.db.commit()
            return existing
    
        def push_quote_to_sf(self, quote):
            # Map quote to SF format
            sf_data = self.mapper.local_to_sf(quote, "quote")
    
            if quote.sf_id:
                # Update existing
                self.sf.update_quote(quote.sf_id, sf_data)
            else:
                # Create new
                result = self.sf.create_quote(sf_data)
                quote.sf_id = result["id"]
                self.db.commit()
    
            return quote.sf_id
    

Learning milestones:

  1. OAuth authentication works → You understand Salesforce auth
  2. Read opportunities → You understand data retrieval
  3. Create quotes in SF → You understand data creation
  4. PDF attachments work → You understand file handling
  5. Bidirectional sync works → Full integration complete

Project 8: Product Configuration DSL

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: Python (with PLY or Lark)
  • Alternative Programming Languages: TypeScript, Rust, Go
  • Coolness Level: Level 4: Hardcore Tech Flex
  • Business Potential: 4. The “Open Core” Infrastructure
  • Difficulty: Level 4: Expert
  • Knowledge Area: Language Design / Parsing / DSLs
  • Software or Tool: Lark Parser, ANTLR, PLY
  • Main Book: “Language Implementation Patterns” by Terence Parr

What you’ll build: A domain-specific language for defining product configurations, rules, and constraints that business analysts (non-programmers) can read and write—transforming complex nested JSON into human-readable configuration files.

Why it teaches DSL design: Real CPQ systems need rules that business users can manage without developer involvement. A well-designed DSL bridges the gap between technical implementation and business requirements. This project forces you to think about syntax design, error messages, and the trade-off between expressiveness and simplicity.

Core challenges you’ll face:

  • Syntax design → maps to human-readable vs machine-parseable trade-offs
  • Parser implementation → maps to lexing, parsing, AST construction
  • Semantic analysis → maps to type checking, reference resolution
  • Error messages → maps to source locations, helpful diagnostics
  • Integration → maps to compiling DSL to runtime rules

Key Concepts:

  • DSL Design: “Domain Specific Languages” Chapter 2-3 - Martin Fowler
  • Parsing Techniques: “Language Implementation Patterns” Chapter 2-5 - Terence Parr
  • Lark Parser: Lark Documentation
  • Grammar Design: “Compilers: Principles and Practice” Chapter 4 - Dave & Dave

Difficulty: Expert Time estimate: 3-4 weeks Prerequisites: Project 2 (Rules Engine) completed, understanding of parsing basics

Real world outcome:

# product_rules.cpq - Human-readable product configuration

product "Enterprise Software License" {
    sku: "ENT-001"
    base_price: $10,000/year

    # Define configurable options
    options {
        user_tier: one_of ["Small (1-50)", "Medium (51-500)", "Large (500+)"]

        modules: many_of [
            "Analytics" @ $3,000/year,
            "API Access" @ $5,000/year,
            "SSO Integration" @ $2,000/year,
            "Advanced Security" @ $4,000/year
        ]

        support_level: one_of ["Standard", "Premium 8x5", "Premium 24x7"]
    }

    # Define rules
    rules {
        # Dependencies
        when "SSO Integration" selected {
            require "API Access"
            message "SSO Integration requires API Access module"
        }

        # Constraints based on selections
        when user_tier is "Large (500+)" {
            require support_level in ["Premium 8x5", "Premium 24x7"]
            message "Enterprise tier requires Premium support"
        }

        # Exclusions
        when user_tier is "Small (1-50)" {
            exclude "Advanced Security"
            message "Advanced Security is only available for Medium and Large tiers"
        }

        # Bundle discounts
        when count(modules) >= 3 {
            recommend "Bundle Discount"
            apply discount 10% to modules
            message "You qualify for our module bundle discount!"
        }
    }

    # Pricing modifiers
    pricing {
        user_tier "Small (1-50)": +$5,000
        user_tier "Medium (51-500)": +$15,000
        user_tier "Large (500+)": +$30,000

        support_level "Premium 8x5": +$2,000
        support_level "Premium 24x7": +$5,000
    }
}

Implementation Hints:

  1. Grammar Definition (Lark):
    # cpq_grammar.lark
    CPQ_GRAMMAR = r"""
    start: product+
    
    product: "product" STRING "{" product_body "}"
    product_body: (sku | base_price | options | rules | pricing)*
    
    sku: "sku:" STRING
    base_price: "base_price:" price
    
    // Options section
    options: "options" "{" option_def+ "}"
    option_def: IDENTIFIER ":" option_type
    option_type: one_of | many_of
    one_of: "one_of" "[" option_list "]"
    many_of: "many_of" "[" option_list "]"
    option_list: option_item ("," option_item)*
    option_item: STRING ("@" price)?
    
    // Rules section
    rules: "rules" "{" rule_def* "}"
    rule_def: "when" condition "{" action+ "}"
    
    condition: selection_condition | comparison_condition | count_condition
    selection_condition: STRING "selected"
    comparison_condition: IDENTIFIER "is" STRING
    count_condition: "count(" IDENTIFIER ")" comparison_op NUMBER
    
    action: require_action | exclude_action | recommend_action | apply_action | message_action
    require_action: "require" (STRING | IDENTIFIER "in" string_list)
    exclude_action: "exclude" STRING
    recommend_action: "recommend" STRING
    apply_action: "apply" "discount" PERCENT "to" IDENTIFIER
    message_action: "message" STRING
    
    // Pricing section
    pricing: "pricing" "{" price_modifier+ "}"
    price_modifier: IDENTIFIER STRING ":" price_delta
    
    // Primitives
    price: "$" NUMBER ("/" TIME_UNIT)?
    price_delta: ("+" | "-") price
    comparison_op: ">=" | "<=" | ">" | "<" | "=="
    string_list: "[" STRING ("," STRING)* "]"
    
    TIME_UNIT: "year" | "month" | "day"
    PERCENT: NUMBER "%"
    IDENTIFIER: /[a-z_][a-z0-9_]*/
    STRING: /"[^"]*"/
    NUMBER: /\d+(\.\d+)?/
    COMMENT: /#[^\n]*/
    
    %import common.WS
    %ignore WS
    %ignore COMMENT
    """
    
  2. AST Classes:
    from dataclasses import dataclass, field
    from typing import List, Optional, Union
    from decimal import Decimal
    from enum import Enum
    
    class OptionType(Enum):
        ONE_OF = "one_of"
        MANY_OF = "many_of"
    
    @dataclass
    class Price:
        amount: Decimal
        period: Optional[str] = None  # year, month, None for one-time
    
    @dataclass
    class OptionItem:
        label: str
        price: Optional[Price] = None
    
    @dataclass
    class OptionDef:
        name: str
        type: OptionType
        items: List[OptionItem]
    
    @dataclass
    class Condition:
        pass
    
    @dataclass
    class SelectionCondition(Condition):
        option_name: str
    
    @dataclass
    class ComparisonCondition(Condition):
        field: str
        operator: str
        value: Union[str, Decimal]
    
    @dataclass
    class CountCondition(Condition):
        field: str
        operator: str
        value: int
    
    @dataclass
    class Action:
        pass
    
    @dataclass
    class RequireAction(Action):
        target: Union[str, List[str]]
        field: Optional[str] = None  # For "field in [list]" syntax
    
    @dataclass
    class ExcludeAction(Action):
        target: str
    
    @dataclass
    class ApplyDiscountAction(Action):
        percent: Decimal
        target: str
    
    @dataclass
    class MessageAction(Action):
        text: str
    
    @dataclass
    class Rule:
        condition: Condition
        actions: List[Action]
        source_line: int  # For error reporting
    
    @dataclass
    class PriceModifier:
        option_name: str
        option_value: str
        delta: Price
    
    @dataclass
    class Product:
        name: str
        sku: str
        base_price: Price
        options: List[OptionDef] = field(default_factory=list)
        rules: List[Rule] = field(default_factory=list)
        pricing: List[PriceModifier] = field(default_factory=list)
    
  3. Parser & Transformer:
    from lark import Lark, Transformer, v_args
    
    class CPQTransformer(Transformer):
        def start(self, items):
            return items
    
        def product(self, items):
            name = items[0]
            body = items[1:]
    
            product = Product(name=name.strip('"'), sku="", base_price=Price(Decimal(0)))
    
            for item in body:
                if isinstance(item, dict):
                    if "sku" in item:
                        product.sku = item["sku"]
                    elif "base_price" in item:
                        product.base_price = item["base_price"]
                    elif "options" in item:
                        product.options = item["options"]
                    elif "rules" in item:
                        product.rules = item["rules"]
                    elif "pricing" in item:
                        product.pricing = item["pricing"]
    
            return product
    
        def rule_def(self, items):
            condition = items[0]
            actions = items[1:]
            return Rule(condition=condition, actions=actions, source_line=0)
    
        @v_args(inline=True)
        def selection_condition(self, option_name):
            return SelectionCondition(option_name=option_name.strip('"'))
    
        @v_args(inline=True)
        def comparison_condition(self, field, value):
            return ComparisonCondition(
                field=str(field),
                operator="==",
                value=value.strip('"')
            )
    
        @v_args(inline=True)
        def require_action(self, target):
            if isinstance(target, list):
                return RequireAction(target=target)
            return RequireAction(target=target.strip('"'))
    
        # ... more transformation methods
    
    class CPQParser:
        def __init__(self):
            self.parser = Lark(CPQ_GRAMMAR, parser='lalr', transformer=CPQTransformer())
    
        def parse(self, source: str) -> List[Product]:
            try:
                return self.parser.parse(source)
            except Exception as e:
                raise CPQSyntaxError(str(e))
    
        def parse_file(self, filepath: str) -> List[Product]:
            with open(filepath) as f:
                return self.parse(f.read())
    
  4. Semantic Analyzer:
    class SemanticAnalyzer:
        def __init__(self):
            self.errors: List[SemanticError] = []
            self.warnings: List[SemanticWarning] = []
    
        def analyze(self, products: List[Product]) -> bool:
            for product in products:
                self._analyze_product(product)
            return len(self.errors) == 0
    
        def _analyze_product(self, product: Product):
            # Collect all valid option names and values
            valid_options = {}
            valid_values = set()
    
            for opt in product.options:
                valid_options[opt.name] = opt
                for item in opt.items:
                    valid_values.add(item.label)
    
            # Check rules reference valid options
            for rule in product.rules:
                self._check_condition(rule.condition, valid_options, valid_values, rule.source_line)
                for action in rule.actions:
                    self._check_action(action, valid_options, valid_values, rule.source_line)
    
            # Check pricing references valid options
            for modifier in product.pricing:
                if modifier.option_name not in valid_options:
                    self.errors.append(SemanticError(
                        f"Unknown option '{modifier.option_name}' in pricing",
                        line=0
                    ))
    
        def _check_condition(self, condition, valid_options, valid_values, line):
            if isinstance(condition, SelectionCondition):
                if condition.option_name not in valid_values:
                    self.errors.append(SemanticError(
                        f"Unknown option value '{condition.option_name}'",
                        line=line
                    ))
            elif isinstance(condition, ComparisonCondition):
                if condition.field not in valid_options:
                    self.errors.append(SemanticError(
                        f"Unknown option '{condition.field}'",
                        line=line
                    ))
    
        def _check_action(self, action, valid_options, valid_values, line):
            if isinstance(action, RequireAction):
                if isinstance(action.target, str) and action.target not in valid_values:
                    self.warnings.append(SemanticWarning(
                        f"Required item '{action.target}' not found in options",
                        line=line
                    ))
    
  5. Compiler to Runtime Rules:
    class CPQCompiler:
        """Compiles DSL AST to runtime rules engine format."""
    
        def compile(self, products: List[Product]) -> Dict:
            """Compile products to JSON rules for the rules engine."""
            result = {"products": [], "rules": []}
    
            for product in products:
                # Compile product definition
                compiled_product = {
                    "sku": product.sku,
                    "name": product.name,
                    "base_price": float(product.base_price.amount),
                    "attributes": []
                }
    
                for opt in product.options:
                    compiled_product["attributes"].append({
                        "name": opt.name,
                        "type": "select" if opt.type == OptionType.ONE_OF else "multiselect",
                        "options": [
                            {"value": item.label, "price_modifier": float(item.price.amount) if item.price else 0}
                            for item in opt.items
                        ]
                    })
    
                result["products"].append(compiled_product)
    
                # Compile rules
                for rule in product.rules:
                    compiled_rule = self._compile_rule(rule, product)
                    result["rules"].append(compiled_rule)
    
            return result
    
        def _compile_rule(self, rule: Rule, product: Product) -> Dict:
            return {
                "product_id": product.sku,
                "condition": self._compile_condition(rule.condition),
                "actions": [self._compile_action(a) for a in rule.actions]
            }
    
        def _compile_condition(self, condition: Condition) -> Dict:
            if isinstance(condition, SelectionCondition):
                return {"selected": condition.option_name}
            elif isinstance(condition, ComparisonCondition):
                return {condition.field: condition.value}
            elif isinstance(condition, CountCondition):
                return {"selected_count": {condition.field: {condition.operator: condition.value}}}
    
        def _compile_action(self, action: Action) -> Dict:
            if isinstance(action, RequireAction):
                return {"type": "require", "target": action.target}
            elif isinstance(action, ExcludeAction):
                return {"type": "exclude", "target": action.target}
            elif isinstance(action, ApplyDiscountAction):
                return {"type": "discount", "percent": float(action.percent), "target": action.target}
            elif isinstance(action, MessageAction):
                return {"type": "message", "text": action.text}
    
  6. Error Reporting:
    class CPQError:
        def format_error(self, source: str, error: SemanticError) -> str:
            lines = source.split('\n')
            line = lines[error.line - 1] if error.line <= len(lines) else ""
    
            return f"""
    Error at line {error.line}:
      {line}
      {"^" * len(line.strip())}
    
    {error.message}
    
    Hint: {self._get_hint(error)}
    """
    
        def _get_hint(self, error: SemanticError) -> str:
            if "Unknown option" in error.message:
                return "Check that the option name matches one defined in the 'options' section"
            # ... more hints
    

Learning milestones:

  1. Grammar parses valid input → You understand parsing fundamentals
  2. AST correctly represents structure → You understand tree representation
  3. Semantic analysis catches errors → You understand validation
  4. Compiler generates runtime rules → You understand code generation
  5. Good error messages → Ready for real users

Project 9: Pricing Rules DSL

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: Python (with Lark)
  • Alternative Programming Languages: TypeScript, Kotlin, Scala
  • Coolness Level: Level 4: Hardcore Tech Flex
  • Business Potential: 4. The “Open Core” Infrastructure
  • Difficulty: Level 4: Expert
  • Knowledge Area: DSL Design / Expression Evaluation / Business Rules
  • Software or Tool: Lark Parser, expression evaluators
  • Main Book: “Domain Specific Languages” by Martin Fowler

What you’ll build: A dedicated DSL for expressing complex pricing rules, discount calculations, and margin constraints—allowing pricing analysts to define rules without writing code.

Why it teaches expression languages: Pricing logic is often the most complex part of CPQ. A pricing DSL must support arithmetic, conditionals, aggregations, and time-based logic while remaining safe (no arbitrary code execution) and auditable.

Core challenges you’ll face:

  • Expression evaluation → maps to safe arithmetic, variable binding
  • Aggregation functions → maps to sum, count, min, max over line items
  • Temporal logic → maps to date ranges, promotional periods
  • Safety → maps to sandboxing, preventing infinite loops
  • Auditability → maps to tracing which rules applied

Key Concepts:

  • Expression Languages: “Domain Specific Languages” Chapter 20-25 - Martin Fowler
  • Safe Evaluation: Restricted Python
  • Business Rules: “Business Rules Management” - various resources

Difficulty: Expert Time estimate: 3-4 weeks Prerequisites: Project 3 (Pricing Engine) and Project 8 (Configuration DSL) completed

Real world outcome:

# pricing_rules.price - Pricing rules for Enterprise Software

price_book "Standard 2024" {
    effective: from 2024-01-01 to 2024-12-31
    currency: USD

    # Base prices
    prices {
        "ENT-001": $10,000/year
        "ENT-001-USERS-S": $5,000/year
        "ENT-001-USERS-M": $15,000/year
        "ENT-001-USERS-L": $30,000/year
        "MOD-ANALYTICS": $3,000/year
        "MOD-API": $5,000/year
        "MOD-SSO": $2,000/year
        "MOD-SECURITY": $4,000/year
    }
}

discount_rules "Volume Discounts" {
    # Tiered discounts based on order total
    rule "Volume Tier 1" {
        when subtotal between $25,000 and $50,000
        apply 5% to order
        stackable: true
        priority: 100
    }

    rule "Volume Tier 2" {
        when subtotal between $50,001 and $100,000
        apply 10% to order
        stackable: true
        priority: 100
    }

    rule "Volume Tier 3" {
        when subtotal > $100,000
        apply 15% to order
        stackable: true
        priority: 100
    }
}

discount_rules "Customer Segment Discounts" {
    rule "Gold Partner" {
        when customer.segment == "gold_partner"
        apply 5% to order
        stackable: true
        priority: 90
    }

    rule "Strategic Account" {
        when customer.annual_revenue > $1,000,000
        apply 8% to order
        stackable: false  # Doesn't stack with Gold Partner
        priority: 85
    }
}

discount_rules "Bundle Discounts" {
    rule "Module Bundle 3+" {
        when count(items where sku starts_with "MOD-") >= 3
        apply 10% to items where sku starts_with "MOD-"
        message: "Module bundle discount applied!"
        stackable: true
        priority: 80
    }

    rule "Complete Suite" {
        when all_of ["MOD-ANALYTICS", "MOD-API", "MOD-SSO", "MOD-SECURITY"] in items
        apply 20% to items where sku starts_with "MOD-"
        message: "Complete suite discount - maximum savings!"
        stackable: false
        priority: 75
    }
}

discount_rules "Promotional" {
    rule "Q4 Special" {
        when today between 2024-10-01 and 2024-12-31
        and subtotal > $25,000
        apply $5,000 off order
        stackable: true
        priority: 50
        promo_code: "Q4SPECIAL"
    }

    rule "New Customer Welcome" {
        when customer.is_new == true
        and subtotal > $10,000
        apply 15% to first_year
        max_discount: $10,000
        stackable: true
        priority: 60
    }
}

approval_thresholds {
    when total_discount_percent > 25%
        require approval from "vp_sales"
        reason: "Discount exceeds 25%"

    when total_discount_percent > 15%
        require approval from "sales_manager"
        reason: "Discount exceeds 15%"

    when margin_percent < 40%
        require approval from "finance"
        reason: "Margin below 40%"

    when any_manual_discount
        require approval from "sales_manager"
        reason: "Manual discount applied"
}

# Custom pricing formula (for advanced use)
formula "Per-User Pricing" {
    input: user_count (number)

    tiers {
        1..50:     $100/user
        51..200:   $85/user
        201..500:  $70/user
        501+:      $50/user
    }

    # Progressive tier calculation
    calculate: sum(
        min(user_count, 50) * $100 +
        max(0, min(user_count - 50, 150)) * $85 +
        max(0, min(user_count - 200, 300)) * $70 +
        max(0, user_count - 500) * $50
    )
}

Implementation Hints:

  1. Pricing DSL Grammar:
    PRICING_GRAMMAR = r"""
    start: (price_book | discount_rules | approval_thresholds | formula)+
    
    // Price Book
    price_book: "price_book" STRING "{" price_book_body "}"
    price_book_body: (effective_clause | currency_clause | prices_block)*
    effective_clause: "effective:" "from" DATE "to" DATE
    currency_clause: "currency:" CURRENCY_CODE
    prices_block: "prices" "{" price_entry+ "}"
    price_entry: STRING ":" price
    
    // Discount Rules
    discount_rules: "discount_rules" STRING "{" rule+ "}"
    rule: "rule" STRING "{" rule_body "}"
    rule_body: when_clause apply_clause rule_option*
    
    when_clause: "when" condition ("and" condition)*
    condition: comparison | between_expr | count_expr | all_of_expr | exists_expr | date_expr
    
    comparison: field_ref COMP_OP value
    between_expr: field_ref "between" value "and" value
    count_expr: "count(" item_filter ")" COMP_OP NUMBER
    all_of_expr: "all_of" string_list "in" IDENTIFIER
    exists_expr: field_ref "in" IDENTIFIER
    date_expr: "today" "between" DATE "and" DATE
    
    apply_clause: "apply" discount_value "to" discount_target
    discount_value: PERCENT | price "off"
    discount_target: "order" | "first_year" | "items" item_filter?
    item_filter: "where" filter_condition
    filter_condition: IDENTIFIER ("starts_with" | "ends_with" | "contains" | "==") STRING
    
    rule_option: stackable_opt | priority_opt | message_opt | max_discount_opt | promo_opt
    stackable_opt: "stackable:" BOOL
    priority_opt: "priority:" NUMBER
    message_opt: "message:" STRING
    max_discount_opt: "max_discount:" price
    promo_opt: "promo_code:" STRING
    
    // Approval Thresholds
    approval_thresholds: "approval_thresholds" "{" threshold+ "}"
    threshold: "when" threshold_condition require_clause reason_clause
    threshold_condition: field_ref COMP_OP value | "any_manual_discount"
    require_clause: "require" "approval" "from" STRING
    reason_clause: "reason:" STRING
    
    // Custom Formula
    formula: "formula" STRING "{" formula_body "}"
    formula_body: input_clause tiers_block? calculate_clause
    input_clause: "input:" IDENTIFIER "(" TYPE ")"
    tiers_block: "tiers" "{" tier_entry+ "}"
    tier_entry: range ":" price_per_unit
    range: NUMBER ".." (NUMBER | "+")
    price_per_unit: price "/" IDENTIFIER
    calculate_clause: "calculate:" expression
    
    // Expressions
    expression: term ((PLUS | MINUS) term)*
    term: factor ((MULT | DIV) factor)*
    factor: function_call | field_ref | NUMBER | price | "(" expression ")"
    function_call: FUNC_NAME "(" (expression ("," expression)*)? ")"
    FUNC_NAME: "sum" | "min" | "max" | "count" | "avg"
    
    // Field references
    field_ref: IDENTIFIER ("." IDENTIFIER)*
    
    // Primitives
    price: "$" NUMBER ("/" TIME_UNIT)?
    value: price | NUMBER | STRING | BOOL
    COMP_OP: ">" | "<" | ">=" | "<=" | "==" | "!="
    PLUS: "+"
    MINUS: "-"
    MULT: "*"
    DIV: "/"
    DATE: /\d{4}-\d{2}-\d{2}/
    CURRENCY_CODE: /[A-Z]{3}/
    TYPE: "number" | "string" | "date"
    BOOL: "true" | "false"
    PERCENT: NUMBER "%"
    TIME_UNIT: "year" | "month" | "user" | "unit"
    
    %import common.SIGNED_NUMBER -> NUMBER
    %import common.ESCAPED_STRING -> STRING
    %import common.CNAME -> IDENTIFIER
    %import common.WS
    %ignore WS
    """
    
  2. Safe Expression Evaluator:
    from decimal import Decimal
    from typing import Dict, Any, Callable
    
    class SafeExpressionEvaluator:
        """Evaluates pricing expressions without arbitrary code execution."""
    
        MAX_ITERATIONS = 10000
        MAX_RECURSION = 50
    
        def __init__(self):
            self.functions: Dict[str, Callable] = {
                "sum": lambda items: sum(items),
                "min": lambda items: min(items) if items else Decimal(0),
                "max": lambda items: max(items) if items else Decimal(0),
                "count": lambda items: len(items),
                "avg": lambda items: sum(items) / len(items) if items else Decimal(0),
            }
            self.iteration_count = 0
    
        def evaluate(self, expr: Expression, context: Dict[str, Any]) -> Decimal:
            self.iteration_count = 0
            return self._eval(expr, context, depth=0)
    
        def _eval(self, expr: Expression, context: Dict[str, Any], depth: int) -> Decimal:
            if depth > self.MAX_RECURSION:
                raise EvaluationError("Maximum recursion depth exceeded")
    
            self.iteration_count += 1
            if self.iteration_count > self.MAX_ITERATIONS:
                raise EvaluationError("Maximum iterations exceeded")
    
            if isinstance(expr, NumberLiteral):
                return Decimal(str(expr.value))
    
            elif isinstance(expr, PriceLiteral):
                return expr.amount
    
            elif isinstance(expr, FieldReference):
                return self._resolve_field(expr.path, context)
    
            elif isinstance(expr, BinaryOp):
                left = self._eval(expr.left, context, depth + 1)
                right = self._eval(expr.right, context, depth + 1)
    
                if expr.op == '+':
                    return left + right
                elif expr.op == '-':
                    return left - right
                elif expr.op == '*':
                    return left * right
                elif expr.op == '/':
                    if right == 0:
                        raise EvaluationError("Division by zero")
                    return left / right
    
            elif isinstance(expr, FunctionCall):
                if expr.name not in self.functions:
                    raise EvaluationError(f"Unknown function: {expr.name}")
    
                args = [self._eval(arg, context, depth + 1) for arg in expr.args]
                return self.functions[expr.name](args)
    
            raise EvaluationError(f"Unknown expression type: {type(expr)}")
    
        def _resolve_field(self, path: List[str], context: Dict) -> Decimal:
            current = context
            for part in path:
                if isinstance(current, dict):
                    current = current.get(part)
                elif hasattr(current, part):
                    current = getattr(current, part)
                else:
                    raise EvaluationError(f"Cannot resolve field: {'.'.join(path)}")
    
                if current is None:
                    return Decimal(0)
    
            return Decimal(str(current)) if current is not None else Decimal(0)
    
  3. Discount Rule Evaluator:
    @dataclass
    class DiscountResult:
        rule_name: str
        discount_amount: Decimal
        discount_percent: Decimal
        applies_to: str  # "order", "items", "line_item_ids"
        line_item_ids: Optional[List[str]] = None
        message: Optional[str] = None
        stackable: bool = True
    
    class DiscountRuleEvaluator:
        def __init__(self, rules: List[DiscountRuleSet]):
            self.rule_sets = rules
            self.expression_eval = SafeExpressionEvaluator()
    
        def evaluate_discounts(self, context: PricingContext) -> List[DiscountResult]:
            """Evaluate all discount rules and return applicable discounts."""
            results = []
    
            # Flatten all rules from all rule sets
            all_rules = []
            for rule_set in self.rule_sets:
                all_rules.extend(rule_set.rules)
    
            # Sort by priority (higher first)
            all_rules.sort(key=lambda r: r.priority, reverse=True)
    
            for rule in all_rules:
                if self._evaluate_condition(rule.when_clause, context):
                    result = self._apply_discount(rule, context)
                    if result:
                        results.append(result)
    
            return self._resolve_stacking(results)
    
        def _evaluate_condition(self, conditions: List[Condition], context: PricingContext) -> bool:
            """All conditions must be true (AND logic)."""
            for condition in conditions:
                if not self._eval_single_condition(condition, context):
                    return False
            return True
    
        def _eval_single_condition(self, condition: Condition, context: PricingContext) -> bool:
            if isinstance(condition, Comparison):
                left = self._get_value(condition.field, context)
                right = condition.value
                return self._compare(left, condition.operator, right)
    
            elif isinstance(condition, BetweenExpr):
                value = self._get_value(condition.field, context)
                return condition.low <= value <= condition.high
    
            elif isinstance(condition, CountExpr):
                items = self._filter_items(context.items, condition.filter)
                count = len(items)
                return self._compare(count, condition.operator, condition.value)
    
            elif isinstance(condition, AllOfExpr):
                item_skus = {item.sku for item in context.items}
                return all(sku in item_skus for sku in condition.required_skus)
    
            elif isinstance(condition, DateExpr):
                return condition.start <= context.today <= condition.end
    
            return False
    
        def _apply_discount(self, rule: Rule, context: PricingContext) -> DiscountResult:
            apply_clause = rule.apply_clause
    
            if apply_clause.target == "order":
                base_amount = context.subtotal
            elif apply_clause.target == "first_year":
                base_amount = context.first_year_total
            elif apply_clause.target == "items":
                filtered = self._filter_items(context.items, apply_clause.filter)
                base_amount = sum(item.total for item in filtered)
    
            if apply_clause.is_percent:
                discount_amount = base_amount * apply_clause.value / 100
                discount_percent = apply_clause.value
            else:
                discount_amount = min(apply_clause.value, base_amount)
                discount_percent = (discount_amount / base_amount * 100) if base_amount else Decimal(0)
    
            # Apply max discount cap if specified
            if rule.max_discount and discount_amount > rule.max_discount:
                discount_amount = rule.max_discount
                discount_percent = (discount_amount / base_amount * 100) if base_amount else Decimal(0)
    
            return DiscountResult(
                rule_name=rule.name,
                discount_amount=discount_amount,
                discount_percent=discount_percent,
                applies_to=apply_clause.target,
                message=rule.message,
                stackable=rule.stackable
            )
    
        def _resolve_stacking(self, results: List[DiscountResult]) -> List[DiscountResult]:
            """Handle non-stackable discounts - keep best one per category."""
            stackable = [r for r in results if r.stackable]
            non_stackable = [r for r in results if not r.stackable]
    
            if non_stackable:
                # Keep only the best non-stackable discount
                best = max(non_stackable, key=lambda r: r.discount_amount)
                return stackable + [best]
    
            return stackable
    
  4. Audit Trail:
    @dataclass
    class PricingAuditEntry:
        timestamp: datetime
        rule_name: str
        condition_details: Dict[str, Any]
        applied: bool
        discount_amount: Optional[Decimal]
        reason: str
    
    class AuditablePricingEngine:
        def __init__(self, evaluator: DiscountRuleEvaluator):
            self.evaluator = evaluator
            self.audit_log: List[PricingAuditEntry] = []
    
        def calculate_with_audit(self, context: PricingContext) -> Tuple[PricingResult, List[PricingAuditEntry]]:
            self.audit_log = []
    
            # Wrap evaluator to capture audit info
            original_eval = self.evaluator._eval_single_condition
    
            def audited_eval(condition, ctx):
                result = original_eval(condition, ctx)
                self.audit_log.append(PricingAuditEntry(
                    timestamp=datetime.now(),
                    rule_name=getattr(ctx, '_current_rule', 'unknown'),
                    condition_details=self._condition_to_dict(condition),
                    applied=result,
                    discount_amount=None,
                    reason=f"Condition {'matched' if result else 'did not match'}"
                ))
                return result
    
            self.evaluator._eval_single_condition = audited_eval
    
            try:
                result = self.evaluator.evaluate_discounts(context)
                return result, self.audit_log
            finally:
                self.evaluator._eval_single_condition = original_eval
    

Learning milestones:

  1. Parse pricing rules → You understand complex grammars
  2. Safe expression evaluation → You understand sandboxed execution
  3. Discount stacking works → You understand business logic
  4. Audit trail captures decisions → You understand traceability
  5. Approval thresholds trigger → Full pricing DSL complete

Project 10: Visual Rule Builder (No-Code DSL)

  • File: LEARN_CPQ_CONFIGURE_PRICE_QUOTE_DEEP_DIVE.md
  • Main Programming Language: TypeScript (React)
  • Alternative Programming Languages: Vue.js, Svelte
  • Coolness Level: Level 4: Hardcore Tech Flex
  • Business Potential: 5. The “Industry Disruptor”
  • Difficulty: Level 3: Advanced
  • Knowledge Area: Visual Programming / AST Manipulation / UI/UX
  • Software or Tool: React, DnD libraries, Monaco Editor
  • Main Book: “Designing Interfaces” by Jenifer Tidwell

What you’ll build: A visual drag-and-drop interface for building configuration and pricing rules—the “Scratch for CPQ” that lets business users create rules without writing any DSL code, while generating valid DSL under the hood.

Why it teaches visual programming: The ultimate DSL is one you don’t have to type. Visual rule builders democratize rule creation while maintaining the rigor of a formal language. This project combines UI design, AST manipulation, and code generation.

Core challenges you’ll face:

  • Visual AST editing → maps to representing trees as blocks/nodes
  • Drag-and-drop UX → maps to intuitive rule construction
  • Real-time validation → maps to showing errors as users build
  • Two-way sync → maps to visual ↔ text DSL synchronization
  • Code generation → maps to converting visual representation to DSL

Key Concepts:

  • Visual Programming: Blockly, Scratch
  • AST Visualization: “Crafting Interpreters” visualization techniques
  • DnD Libraries: React DnD, dnd-kit
  • Code Generation: “Domain Specific Languages” Chapter 30 - Martin Fowler

Difficulty: Advanced Time estimate: 4-5 weeks Prerequisites: React expertise, Projects 8-9 (DSLs) completed

Real world outcome:

┌─────────────────────────────────────────────────────────────────────────────┐
│  Visual Rule Builder                                        [💾 Save] [▶ Test] │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  ┌─ Components ─┐     ┌─────────────── Rule Canvas ────────────────────┐    │
│  │              │     │                                                 │    │
│  │  📦 Conditions │     │   ┌──────────────────────────────────────┐     │    │
│  │  ├─ When     │     │   │  RULE: SSO Requires API                │     │    │
│  │  ├─ And      │     │   ├──────────────────────────────────────┤     │    │
│  │  ├─ Or       │     │   │                                        │     │    │
│  │  └─ Not      │     │   │   WHEN  ┌──────────────────────────┐  │     │    │
│  │              │     │   │         │ "SSO Integration"        │  │     │    │
│  │  📦 Comparisons│     │   │         │     is selected          │  │     │    │
│  │  ├─ Equals   │     │   │         └──────────────────────────┘  │     │    │
│  │  ├─ Greater  │     │   │                                        │     │    │
│  │  ├─ Less     │     │   │   THEN  ┌──────────────────────────┐  │     │    │
│  │  └─ Between  │     │   │         │ Require                  │  │     │    │
│  │              │     │   │         │   └─ "API Access"        │  │     │    │
│  │  📦 Actions   │     │   │         └──────────────────────────┘  │     │    │
│  │  ├─ Require  │     │   │                                        │     │    │
│  │  ├─ Exclude  │     │   │         ┌──────────────────────────┐  │     │    │
│  │  ├─ Discount │     │   │         │ Show Message             │  │     │    │
│  │  └─ Message  │     │   │         │   "SSO requires API"     │  │     │    │
│  │              │     │   │         └──────────────────────────┘  │     │    │
│  └──────────────┘     │   │                                        │     │    │
│                       │   └──────────────────────────────────────┘     │    │
│                       │                                                 │    │
│                       │   [+ Add Another Rule]                          │    │
│                       └─────────────────────────────────────────────────┘    │
│                                                                              │
├─────────────────────────────────────────────────────────────────────────────┤
│  Generated DSL:                                                     [Copy]  │
│  ┌───────────────────────────────────────────────────────────────────────┐  │
│  │ rule "SSO Requires API" {                                             │  │
│  │     when "SSO Integration" selected {                                 │  │
│  │         require "API Access"                                          │  │
│  │         message "SSO Integration requires API Access module"          │  │
│  │     }                                                                 │  │
│  │ }                                                                     │  │
│  └───────────────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────────┘

Implementation Hints:

  1. Block Type Definitions:
    // Define the visual block types that map to DSL constructs
    
    type BlockType =
      | 'rule'
      | 'when_condition'
      | 'and_condition'
      | 'or_condition'
      | 'comparison'
      | 'selection_check'
      | 'count_check'
      | 'require_action'
      | 'exclude_action'
      | 'discount_action'
      | 'message_action';
    
    interface BlockDefinition {
      type: BlockType;
      label: string;
      icon: string;
      color: string;
      category: 'condition' | 'action' | 'container';
      inputs: InputDefinition[];
      accepts?: BlockType[];  // What blocks can be nested inside
      output?: 'boolean' | 'action' | 'value';  // What this block produces
    }
    
    interface InputDefinition {
      name: string;
      type: 'text' | 'number' | 'select' | 'product_select' | 'block';
      options?: { value: string; label: string }[];
      placeholder?: string;
      required?: boolean;
    }
    
    const BLOCK_DEFINITIONS: BlockDefinition[] = [
      {
        type: 'rule',
        label: 'Rule',
        icon: '📋',
        color: '#4A90D9',
        category: 'container',
        inputs: [
          { name: 'name', type: 'text', placeholder: 'Rule name', required: true }
        ],
        accepts: ['when_condition']
      },
      {
        type: 'when_condition',
        label: 'When',
        icon: '🔍',
        color: '#F5A623',
        category: 'condition',
        inputs: [],
        accepts: ['selection_check', 'comparison', 'count_check', 'and_condition', 'or_condition'],
        output: 'boolean'
      },
      {
        type: 'selection_check',
        label: 'Is Selected',
        icon: '',
        color: '#7ED321',
        category: 'condition',
        inputs: [
          { name: 'option', type: 'product_select', placeholder: 'Select option', required: true }
        ],
        output: 'boolean'
      },
      {
        type: 'require_action',
        label: 'Require',
        icon: '📌',
        color: '#D0021B',
        category: 'action',
        inputs: [
          { name: 'target', type: 'product_select', placeholder: 'Required item', required: true }
        ],
        output: 'action'
      },
      {
        type: 'discount_action',
        label: 'Apply Discount',
        icon: '💰',
        color: '#50E3C2',
        category: 'action',
        inputs: [
          { name: 'percent', type: 'number', placeholder: 'Discount %', required: true },
          { name: 'target', type: 'select', options: [
            { value: 'order', label: 'Entire Order' },
            { value: 'item', label: 'Specific Items' }
          ]}
        ],
        output: 'action'
      },
      // ... more block definitions
    ];
    
  2. Block State & AST:
    interface Block {
      id: string;
      type: BlockType;
      inputs: Record<string, any>;
      children: Block[];
      parent?: string;
      position?: { x: number; y: number };
    }
    
    interface RuleBuilderState {
      blocks: Record<string, Block>;
      rootBlocks: string[];  // Top-level rule blocks
      selectedBlock: string | null;
      draggedBlock: string | null;
      errors: ValidationError[];
    }
    
    // Convert visual blocks to AST
    function blocksToAST(state: RuleBuilderState): RuleAST[] {
      return state.rootBlocks.map(id => blockToASTNode(state.blocks[id], state.blocks));
    }
    
    function blockToASTNode(block: Block, allBlocks: Record<string, Block>): ASTNode {
      const children = block.children.map(childId =>
        blockToASTNode(allBlocks[childId], allBlocks)
      );
    
      switch (block.type) {
        case 'rule':
          return {
            type: 'Rule',
            name: block.inputs.name,
            conditions: children.filter(c => c.type.includes('Condition')),
            actions: children.filter(c => c.type.includes('Action'))
          };
    
        case 'selection_check':
          return {
            type: 'SelectionCondition',
            optionName: block.inputs.option
          };
    
        case 'require_action':
          return {
            type: 'RequireAction',
            target: block.inputs.target
          };
    
        // ... more cases
      }
    }
    
  3. Drag and Drop System:
    import { useDrag, useDrop } from 'react-dnd';
    
    interface DraggableBlockProps {
      block: Block;
      definition: BlockDefinition;
      onDrop: (source: string, target: string, position: 'before' | 'after' | 'inside') => void;
    }
    
    const DraggableBlock: React.FC<DraggableBlockProps> = ({ block, definition, onDrop }) => {
      const [{ isDragging }, dragRef] = useDrag({
        type: 'BLOCK',
        item: { id: block.id, type: block.type },
        collect: (monitor) => ({
          isDragging: monitor.isDragging()
        })
      });
    
      const [{ isOver, canDrop }, dropRef] = useDrop({
        accept: 'BLOCK',
        canDrop: (item: { id: string; type: BlockType }) => {
          // Check if this block type can accept the dragged block
          if (!definition.accepts) return false;
          return definition.accepts.includes(item.type);
        },
        drop: (item: { id: string }) => {
          onDrop(item.id, block.id, 'inside');
        },
        collect: (monitor) => ({
          isOver: monitor.isOver(),
          canDrop: monitor.canDrop()
        })
      });
    
      return (
        <div
          ref={(node) => dragRef(dropRef(node))}
          className={cn(
            'block',
            `block--${definition.category}`,
            isDragging && 'block--dragging',
            isOver && canDrop && 'block--drop-target'
          )}
          style={{ backgroundColor: definition.color }}
        >
          <div className="block__header">
            <span className="block__icon">{definition.icon}</span>
            <span className="block__label">{definition.label}</span>
          </div>
    
          <div className="block__inputs">
            {definition.inputs.map(input => (
              <BlockInput
                key={input.name}
                input={input}
                value={block.inputs[input.name]}
                onChange={(value) => updateBlockInput(block.id, input.name, value)}
              />
            ))}
          </div>
    
          {definition.accepts && (
            <div className="block__children">
              {block.children.map(childId => (
                <DraggableBlock
                  key={childId}
                  block={blocks[childId]}
                  definition={getDefinition(blocks[childId].type)}
                  onDrop={onDrop}
                />
              ))}
              <DropPlaceholder
                accepts={definition.accepts}
                onDrop={(sourceId) => onDrop(sourceId, block.id, 'inside')}
              />
            </div>
          )}
        </div>
      );
    };
    
  4. DSL Code Generator:
    class DSLCodeGenerator {
      private indentLevel = 0;
      private output: string[] = [];
    
      generate(ast: RuleAST[]): string {
        this.output = [];
        this.indentLevel = 0;
    
        for (const rule of ast) {
          this.generateRule(rule);
          this.output.push('');
        }
    
        return this.output.join('\n');
      }
    
      private generateRule(rule: RuleAST) {
        this.line(`rule "${rule.name}" {`);
        this.indent();
    
        // Generate conditions
        if (rule.conditions.length > 0) {
          this.line('when ' + this.generateCondition(rule.conditions[0]) + ' {');
          this.indent();
    
          // Generate actions
          for (const action of rule.actions) {
            this.generateAction(action);
          }
    
          this.dedent();
          this.line('}');
        }
    
        this.dedent();
        this.line('}');
      }
    
      private generateCondition(condition: ConditionAST): string {
        switch (condition.type) {
          case 'SelectionCondition':
            return `"${condition.optionName}" selected`;
    
          case 'ComparisonCondition':
            return `${condition.field} ${condition.operator} ${this.formatValue(condition.value)}`;
    
          case 'AndCondition':
            return condition.conditions.map(c => this.generateCondition(c)).join('\n    and ');
    
          case 'OrCondition':
            return '(' + condition.conditions.map(c => this.generateCondition(c)).join(' or ') + ')';
    
          case 'CountCondition':
            return `count(${condition.field}) ${condition.operator} ${condition.value}`;
        }
      }
    
      private generateAction(action: ActionAST) {
        switch (action.type) {
          case 'RequireAction':
            this.line(`require "${action.target}"`);
            break;
    
          case 'ExcludeAction':
            this.line(`exclude "${action.target}"`);
            break;
    
          case 'DiscountAction':
            this.line(`apply ${action.percent}% to ${action.target}`);
            break;
    
          case 'MessageAction':
            this.line(`message "${action.text}"`);
            break;
        }
      }
    
      private line(text: string) {
        this.output.push('  '.repeat(this.indentLevel) + text);
      }
    
      private indent() { this.indentLevel++; }
      private dedent() { this.indentLevel--; }
    
      private formatValue(value: any): string {
        if (typeof value === 'string') return `"${value}"`;
        if (typeof value === 'number') return value.toString();
        return String(value);
      }
    }
    
  5. Two-Way Sync (Visual ↔ Text):
    class TwoWaySync {
      private parser: CPQParser;
      private generator: DSLCodeGenerator;
      private blockBuilder: BlockBuilder;
    
      constructor() {
        this.parser = new CPQParser();
        this.generator = new DSLCodeGenerator();
        this.blockBuilder = new BlockBuilder();
      }
    
      // Visual → Text
      visualToText(state: RuleBuilderState): string {
        const ast = blocksToAST(state);
        return this.generator.generate(ast);
      }
    
      // Text → Visual
      textToVisual(dslCode: string): RuleBuilderState {
        try {
          const ast = this.parser.parse(dslCode);
          return this.blockBuilder.buildFromAST(ast);
        } catch (error) {
          throw new SyncError('Failed to parse DSL code', error);
        }
      }
    
      // Incremental sync for real-time editing
      syncChanges(
        visualState: RuleBuilderState,
        textCode: string,
        lastEditSource: 'visual' | 'text'
      ): { visual: RuleBuilderState; text: string } {
        if (lastEditSource === 'visual') {
          return {
            visual: visualState,
            text: this.visualToText(visualState)
          };
        } else {
          return {
            visual: this.textToVisual(textCode),
            text: textCode
          };
        }
      }
    }
    
    // BlockBuilder converts AST back to visual blocks
    class BlockBuilder {
      buildFromAST(ast: RuleAST[]): RuleBuilderState {
        const blocks: Record<string, Block> = {};
        const rootBlocks: string[] = [];
    
        for (const rule of ast) {
          const ruleBlock = this.buildRuleBlock(rule, blocks);
          rootBlocks.push(ruleBlock.id);
        }
    
        return {
          blocks,
          rootBlocks,
          selectedBlock: null,
          draggedBlock: null,
          errors: []
        };
      }
    
      private buildRuleBlock(rule: RuleAST, blocks: Record<string, Block>): Block {
        const id = generateId();
        const children: string[] = [];
    
        // Build condition blocks
        for (const condition of rule.conditions) {
          const conditionBlock = this.buildConditionBlock(condition, blocks, id);
          children.push(conditionBlock.id);
        }
    
        // Build action blocks
        for (const action of rule.actions) {
          const actionBlock = this.buildActionBlock(action, blocks, id);
          children.push(actionBlock.id);
        }
    
        const block: Block = {
          id,
          type: 'rule',
          inputs: { name: rule.name },
          children,
          parent: undefined
        };
    
        blocks[id] = block;
        return block;
      }
    
      // ... buildConditionBlock, buildActionBlock implementations
    }
    
  6. Real-Time Validation:
    interface ValidationError {
      blockId: string;
      field?: string;
      message: string;
      severity: 'error' | 'warning';
    }
    
    class RealTimeValidator {
      validate(state: RuleBuilderState, productContext: ProductContext): ValidationError[] {
        const errors: ValidationError[] = [];
    
        for (const [id, block] of Object.entries(state.blocks)) {
          // Check required inputs
          const definition = getBlockDefinition(block.type);
          for (const input of definition.inputs) {
            if (input.required && !block.inputs[input.name]) {
              errors.push({
                blockId: id,
                field: input.name,
                message: `${input.placeholder || input.name} is required`,
                severity: 'error'
              });
            }
          }
    
          // Check product references are valid
          if (block.type === 'selection_check' || block.type === 'require_action') {
            const option = block.inputs.option || block.inputs.target;
            if (option && !productContext.isValidOption(option)) {
              errors.push({
                blockId: id,
                message: `"${option}" is not a valid product option`,
                severity: 'error'
              });
            }
          }
    
          // Check for circular dependencies
          if (block.type === 'require_action') {
            const circularCheck = this.checkCircularDependency(block, state, productContext);
            if (circularCheck) {
              errors.push({
                blockId: id,
                message: circularCheck,
                severity: 'warning'
              });
            }
          }
        }
    
        return errors;
      }
    
      private checkCircularDependency(block: Block, state: RuleBuilderState, context: ProductContext): string | null {
        // Find the "when" condition for this rule
        // Check if the required item would create a circular dependency
        // e.g., A requires B, B requires A
        return null;  // Implementation details...
      }
    }
    

Learning milestones:

  1. Blocks render and are draggable → You understand visual representation
  2. Drag-drop creates valid rules → You understand tree manipulation
  3. DSL code generates correctly → You understand code generation
  4. Two-way sync works → You understand AST transformations
  5. Validation shows errors in real-time → Production-ready UI

Project Comparison Table

Project Difficulty Time Depth Fun Factor
1. Product Catalog Intermediate 1-2 weeks ⭐⭐⭐ ⭐⭐
2. Rules Engine Advanced 2-3 weeks ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐
3. Pricing Engine Advanced 2-3 weeks ⭐⭐⭐⭐⭐ ⭐⭐⭐
4. Quote Generator Intermediate 1-2 weeks ⭐⭐⭐ ⭐⭐⭐
5. Approval Workflow Advanced 2-3 weeks ⭐⭐⭐⭐ ⭐⭐⭐⭐
6. Guided Selling Intermediate 2 weeks ⭐⭐⭐ ⭐⭐⭐⭐
7. CRM Integration Advanced 2-3 weeks ⭐⭐⭐⭐ ⭐⭐⭐
8. Configuration DSL Expert 3-4 weeks ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐
9. Pricing Rules DSL Expert 3-4 weeks ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐
10. Visual Rule Builder Advanced 4-5 weeks ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐

Phase 1: Foundations (Weeks 1-4)

  1. Project 1: Product Catalog - Build the data foundation
  2. Project 2: Rules Engine - Add configuration intelligence

Phase 2: Core CPQ (Weeks 5-8)

  1. Project 3: Pricing Engine - Calculate prices correctly
  2. Project 4: Quote Generator - Create professional output

Phase 3: Enterprise Features (Weeks 9-12)

  1. Project 5: Approval Workflow - Add business controls
  2. Project 6: Guided Selling - Improve user experience
  3. Project 7: CRM Integration - Connect to Salesforce

Phase 4: DSL & No-Code (Weeks 13-20)

  1. Project 8: Configuration DSL - Human-readable rule definitions
  2. Project 9: Pricing Rules DSL - Business analyst-friendly pricing
  3. Project 10: Visual Rule Builder - No-code interface for rules

Final Capstone: Complete CPQ Platform with DSL Studio

Combine all projects into a complete CPQ platform with:

  • Full product catalog with configurable products
  • Real-time configuration validation
  • Dynamic pricing with all discount types
  • Professional quote generation
  • Approval workflows with escalation
  • Guided selling wizard
  • Salesforce integration
  • Configuration DSL for human-readable product definitions
  • Pricing DSL for business analyst-managed pricing rules
  • Visual Rule Builder (“CPQ Studio”) for no-code rule creation
  • Admin UI with Monaco-based DSL editor
  • Live preview and two-way sync between visual and text DSL
  • Analytics dashboard

This capstone would be a legitimate enterprise SaaS product with unique no-code/low-code capabilities that differentiate it from competitors. The DSL approach allows technical users to work in text while business users work visually—both producing the same underlying rules.


Summary

# Project Main Language
1 Product Catalog Python
2 Configuration Rules Engine Python
3 Pricing Engine Python
4 Quote Builder & Document Generator Python
5 Approval Workflow Engine Python
6 Guided Selling Assistant TypeScript (React)
7 CRM Integration (Salesforce) Python
8 Product Configuration DSL Python (Lark)
9 Pricing Rules DSL Python (Lark)
10 Visual Rule Builder (No-Code DSL) TypeScript (React)
Final Complete CPQ Platform with DSL Studio Python + TypeScript

Essential Resources

CPQ Industry

Technical Resources

Books

  • “Domain-Driven Design” by Eric Evans - Data modeling
  • “Enterprise Integration Patterns” by Hohpe & Woolf - Integration
  • “The Strategy and Tactics of Pricing” by Nagle & Holden - Pricing concepts

Master these projects and you’ll understand how enterprise sales software works—and be able to build systems that companies would pay millions for.