Parts Availability Checks in Work Order Routing Pipelines

Dispatching technicians without verifying component availability creates predictable bottlenecks, idle wrench-time, and cascading SLA violations. In modern CMMS architectures, the parts availability check operates as a deterministic decision gate within the work order routing pipeline. It evaluates real-time inventory state before a maintenance request transitions to technician assignment. For facilities managers and reliability engineers, this gate translates into accurate lead-time forecasting and optimized labor utilization. For Python automation developers and integration teams, it demands strict orchestration, resilient API consumption, and explicit routing logic that respects both asset criticality and stock reality. This implementation focuses exclusively on the routing evaluation stage, building upon the foundational data harmonization established in Asset Lookup & Inventory Synchronization without duplicating base synchronization mechanics.

Routing Decision Architecture

The availability check must function as a synchronous evaluation within an otherwise asynchronous dispatch pipeline. Routing decisions are not strictly binary; they operate across a multi-state matrix: ROUTE_IMMEDIATE, ROUTE_PARTIAL, ROUTE_SUBSTITUTE, or DEFER_PROCUREMENT. The pipeline ingests a structured work order payload containing asset identifiers, bill-of-materials (BOM) references, required quantities, and priority classification. It queries inventory systems, evaluates allocation rules, and returns a routing directive alongside a confidence score. This architecture prevents premature dispatch while maintaining pipeline throughput through non-blocking I/O and explicit state transitions.

Implementation Pipeline

1. Standardize the Routing Payload Schema

Define a strict contract for availability requests at pipeline ingress. Enforce validation to reject malformed payloads before they consume API quotas or trigger downstream side effects. Pydantic provides runtime validation, automatic type coercion, and OpenAPI schema generation suitable for CMMS integrations.

from pydantic import BaseModel, Field, validator
from typing import List, Dict
from enum import Enum

class PriorityLevel(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"

class RoutingRequest(BaseModel):
    work_order_id: str = Field(..., min_length=8, max_length=64)
    asset_id: str = Field(..., min_length=6)
    part_skus: List[str] = Field(..., min_items=1, max_items=50)
    required_quantities: Dict[str, int] = Field(..., min_items=1)
    location_id: str = Field(..., description="Target storeroom or site ID")
    priority_level: PriorityLevel = PriorityLevel.STANDARD

    @validator("required_quantities")
    def validate_sku_alignment(cls, v, values):
        if set(v.keys()) != set(values.get("part_skus", [])):
            raise ValueError("SKU keys in quantities must exactly match part_skus")
        return v

2. Generate Deterministic Idempotency Keys

Network retries, message broker redeliveries, and pipeline restarts are inevitable in distributed maintenance environments. Hash the combination of work_order_id and part_skus using a deterministic algorithm, truncate to 16 bytes, and attach the resulting string as an Idempotency-Key header. This guarantees that duplicate requests do not trigger phantom inventory reservations or conflicting routing states.

import hashlib
import struct
from typing import List

def generate_idempotency_key(work_order_id: str, part_skus: List[str]) -> str:
    # Sort SKUs to ensure deterministic hashing regardless of list order
    canonical_input = f"{work_order_id}|{','.join(sorted(part_skus))}".encode("utf-8")
    digest = hashlib.sha256(canonical_input).digest()
    # Truncate to 16 bytes and encode as URL-safe base64
    return struct.unpack("<Q", digest[:8])[0].to_bytes(8, "big").hex()

3. Execute Asynchronous Availability Queries

Route the pipeline to query live inventory endpoints concurrently. The system should invoke Real-time parts availability checks via REST APIs using non-blocking HTTP clients. Leverage Python’s native event loop for concurrent execution without thread contention. Parse responses for on_hand, allocated, in_transit, and available_for_reservation fields. Calculate net availability using net = on_hand - allocated + in_transit.

import httpx
from typing import Dict, Any, List

async def fetch_inventory_availability(
    client: httpx.AsyncClient,
    base_url: str,
    skus: List[str],
    idempotency_key: str
) -> Dict[str, Any]:
    headers = {"Idempotency-Key": idempotency_key, "Accept": "application/json"}
    
    # Batch request to inventory endpoint
    async with client.stream("GET", f"{base_url}/inventory/check", 
                             params={"skus": ",".join(skus)}, headers=headers) as response:
        response.raise_for_status()
        return await response.json()

4. Evaluate Allocation Logic & Routing Directives

Compare net availability against required quantities. Apply routing rules based on asset criticality and maintenance window constraints. If stock is sufficient, emit ROUTE_IMMEDIATE. If partial, evaluate substitution rules or cross-dock transfers. For critical assets missing essential components, trigger Automated Reorder Triggers and set routing state to DEFER_PROCUREMENT.

from typing import Dict, Any

# Reuses RoutingRequest and PriorityLevel from the schema defined above.

def evaluate_routing_directive(
    request: RoutingRequest,
    inventory_data: Dict[str, Any]
) -> Dict[str, Any]:
    routing_states = []
    missing_critical = False

    for sku in request.part_skus:
        required = request.required_quantities[sku]
        stock = inventory_data.get(sku, {})
        net_available = (
            stock.get("on_hand", 0) 
            - stock.get("allocated", 0) 
            + stock.get("in_transit", 0)
        )

        if net_available >= required:
            routing_states.append("FULLY_AVAILABLE")
        elif net_available > 0:
            routing_states.append("PARTIAL")
        else:
            routing_states.append("OUT_OF_STOCK")
            if request.priority_level == PriorityLevel.CRITICAL:
                missing_critical = True

    if missing_critical:
        return {"directive": "DEFER_PROCUREMENT", "confidence": 0.95, "missing_skus": [s for s, st in zip(request.part_skus, routing_states) if st == "OUT_OF_STOCK"]}
    elif "OUT_OF_STOCK" not in routing_states and "PARTIAL" not in routing_states:
        return {"directive": "ROUTE_IMMEDIATE", "confidence": 1.0, "missing_skus": []}
    else:
        return {"directive": "ROUTE_PARTIAL", "confidence": 0.75, "missing_skus": [s for s, st in zip(request.part_skus, routing_states) if st != "FULLY_AVAILABLE"]}

Pipeline Boundaries & Failure Modes

Clear boundaries prevent routing logic from leaking into procurement or execution layers. The availability check must enforce strict timeouts (typically 3–5 seconds) and implement circuit breakers to degrade gracefully when inventory APIs become unresponsive. When live endpoints fail, the pipeline should fall back to a time-bound inventory snapshot rather than blocking the entire dispatch queue. Implement exponential backoff with jitter for transient 5xx responses, and log all routing decisions with structured audit trails for compliance and root-cause analysis.

Once routed, the work order status updates in the CMMS. Technicians may use Barcode & QR Integration at the storeroom to confirm physical allocation before departure, closing the loop between digital routing and physical fulfillment. This deterministic gate ensures that maintenance workflows align with ISO 55001 asset management principles, prioritizing reliability over speed while maintaining operational throughput.