Parts Availability Checks in CMMS Work Order Routing

Parts availability checks are the routing-decision stage of the Asset Lookup & Inventory Synchronization pipeline — the gate that evaluates reconciled inventory state before a maintenance request is allowed to transition to technician assignment.

Dispatching a technician without confirming component availability creates predictable bottlenecks: idle wrench-time at an empty bin, cascading SLA violations, and assets that stay down longer than their mean-time-to-repair predicts. This component closes that gap by treating availability as a synchronous, deterministic decision inside an otherwise asynchronous dispatch flow. For facilities managers and reliability engineers it yields accurate lead-time forecasting and higher labor utilization; for Python automation and CMMS integration teams it demands strict orchestration, resilient API consumption, and explicit routing logic that respects both asset criticality and ground-truth stock. This guide implements that stage end to end — prerequisites, the input/output data contract, a step-by-step async build, a configuration reference, validation checks, and the failure modes you will actually hit in production.

Prerequisites

This component runs as a synchronous gate invoked by the dispatch worker. Before you deploy it, confirm the following are in place.

• Python 3.11+ with pydantic>=2.6 for request validation, httpx>=0.27 for async HTTP against the inventory endpoint, and tenacity>=8.2 for retry orchestration. No dedicated scheduler is required; the gate is called inline by the dispatch consumer driven by the broker described in async batch processing.
• CMMS REST API v1 exposing a read-only availability resource (GET /api/v1/inventory/check) that returns on_hand, allocated, and in_transit per SKU and honors an Idempotency-Key header. The endpoint must serve live state, not a cached projection — see real-time parts availability checks via REST APIs for the cache-busting configuration that guarantees this.
• A resolved part master. Availability math is only meaningful once external SKUs map to canonical CMMS parts and those parts are anchored to the equipment tree from asset hierarchy design.
• Environment variables: CMMS_BASE_URL, CMMS_API_TOKEN (carrying the inventory:read scope), and AVAILABILITY_TIMEOUT_S (default 4). A token missing inventory:read must fail closed at the query step rather than returning a false-positive availability signal.

Architecture and Data Contract

The gate sits between a validated work order and the dispatch router. It never mutates inventory and it never raises a purchase order itself — it reads reconciled stock, evaluates allocation rules, and emits a routing directive. Four boundaries keep the decision honest and stop routing logic from leaking into procurement or execution layers.

• Request boundary: a RoutingRequest enters only after schema validation; a malformed payload is rejected before it consumes an API quota.
• Query boundary: live availability is fetched concurrently over an authenticated REST endpoint with an idempotency key and a hard timeout.
• Decision boundary: deterministic allocation math consumes net availability and required quantities and emits a RoutingDecision. Same input, same output, every time.
• Routing boundary: downstream consumers act on the directive — immediate dispatch, partial dispatch, substitution, or deferral to automated reorder triggers — none of which this component performs.

Routing decisions are not binary. The gate operates across a multi-state matrix — ROUTE_IMMEDIATE, ROUTE_PARTIAL, ROUTE_SUBSTITUTE, or DEFER_PROCUREMENT — and returns the directive alongside a confidence score and the list of short SKUs so the caller can act without re-querying.

The contract across the decision boundary is explicit: the input is a RoutingRequest derived from the canonical work order, and the output is a RoutingDecision. Encoding both with pydantic means a malformed request is rejected at the boundary instead of producing a plausible-but-wrong directive.

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


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


class Directive(str, Enum):
    ROUTE_IMMEDIATE = "ROUTE_IMMEDIATE"
    ROUTE_PARTIAL = "ROUTE_PARTIAL"
    ROUTE_SUBSTITUTE = "ROUTE_SUBSTITUTE"
    DEFER_PROCUREMENT = "DEFER_PROCUREMENT"


class RoutingRequest(BaseModel):
    """Input contract: one availability evaluation for one work order."""
    work_order_id: str = Field(..., min_length=8, max_length=64)
    asset_id: str = Field(..., min_length=6)
    part_skus: List[str] = Field(..., min_length=1, max_length=50)
    required_quantities: Dict[str, int] = Field(...)
    location_id: str = Field(..., description="Target storeroom or site ID")
    priority: Priority = Priority.STANDARD
    escalation_tier: int = Field(0, ge=0, le=3)

    @model_validator(mode="after")
    def validate_sku_alignment(self) -> "RoutingRequest":
        if set(self.required_quantities.keys()) != set(self.part_skus):
            raise ValueError("SKU keys in quantities must exactly match part_skus")
        return self


class RoutingDecision(BaseModel):
    """Output contract: the directive plus the evidence behind it."""
    directive: Directive
    confidence: float = Field(..., ge=0.0, le=1.0)
    short_skus: List[str] = Field(default_factory=list)

Step-by-Step Implementation

1. Define the canonical work order payload

The gate is invoked against a work order, so it imports the same WorkOrderPayload used everywhere on this site rather than redefining its own shape. The SLA fields — priority, requested_completion, and escalation_tier — are what the routing engine reads when it decides whether a part shortage should defer or escalate a job. Schema details live in work order schema standards.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional


@dataclass
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    location_id: str
    priority: Priority = Priority.STANDARD
    requested_completion: Optional[datetime] = None
    escalation_tier: int = 0
    status: str = "open"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


def to_routing_request(wo: WorkOrderPayload) -> RoutingRequest:
    """Narrow the work order down to exactly what the gate needs."""
    return RoutingRequest(
        work_order_id=wo.work_order_id,
        asset_id=wo.asset_id,
        part_skus=wo.part_skus,
        required_quantities=wo.required_quantities,
        location_id=wo.location_id,
        priority=wo.priority,
        escalation_tier=wo.escalation_tier,
    )

2. Generate a deterministic idempotency key

Network retries, broker redeliveries, and pipeline restarts are inevitable in distributed maintenance environments. Hash the combination of work_order_id and the sorted SKU list, then attach the digest as an Idempotency-Key header so duplicate requests can never trigger phantom reservations or conflicting routing states. Sorting the SKUs first makes the key independent of list order.

import hashlib
from typing import List


def generate_idempotency_key(work_order_id: str, part_skus: List[str]) -> str:
    """Order-independent key so a re-sent request is recognised as a duplicate."""
    canonical = f"{work_order_id}|{','.join(sorted(part_skus))}".encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()[:16]

3. Fetch live availability concurrently

Query the inventory endpoint over a non-blocking HTTP client so multiple SKUs resolve in parallel without thread contention. Wrap the call in a hard timeout and let tenacity retry only transient 5xx and transport errors with exponential backoff and jitter — a 4xx is a contract error and must not be retried. Parse on_hand, allocated, and in_transit per SKU; net availability is on_hand - allocated + in_transit.

import httpx
from tenacity import (
    retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type,
)
from typing import Any, Dict, List


class InventoryUnavailable(RuntimeError):
    """Raised when live inventory cannot be reached within the SLA budget."""


@retry(
    retry=retry_if_exception_type((httpx.TransportError, httpx.HTTPStatusError)),
    wait=wait_exponential_jitter(initial=0.2, max=2.0),
    stop=stop_after_attempt(3),
    reraise=True,
)
async def fetch_availability(
    client: httpx.AsyncClient,
    base_url: str,
    skus: List[str],
    idempotency_key: str,
    timeout_s: float = 4.0,
) -> Dict[str, Any]:
    """Return live per-SKU stock; retries only transient failures."""
    response = await client.get(
        f"{base_url.rstrip('/')}/api/v1/inventory/check",
        params={"skus": ",".join(sorted(skus))},
        headers={"Idempotency-Key": idempotency_key, "Accept": "application/json"},
        timeout=timeout_s,
    )
    # 5xx is retryable; raise_for_status surfaces it to tenacity. 4xx reraises out.
    if response.status_code >= 500:
        response.raise_for_status()
    if response.status_code >= 400:
        raise InventoryUnavailable(f"contract error {response.status_code}")
    return response.json()

4. Evaluate allocation logic and emit a directive

Compare net availability against required quantities, then apply routing rules keyed on asset criticality. Full stock yields ROUTE_IMMEDIATE; a critical part at zero stock yields DEFER_PROCUREMENT (which the caller forwards to reorder automation); anything in between yields ROUTE_PARTIAL so the technician can begin the work they can. The math is pure and deterministic, which is what makes the decision testable in isolation.

def evaluate_routing_directive(
    request: RoutingRequest,
    inventory: Dict[str, Any],
) -> RoutingDecision:
    """Deterministic allocation math → one of four routing directives."""
    fully, short, missing_critical = [], [], False

    for sku in request.part_skus:
        required = request.required_quantities[sku]
        stock = inventory.get(sku, {})
        net = (
            stock.get("on_hand", 0)
            - stock.get("allocated", 0)
            + stock.get("in_transit", 0)
        )
        if net >= required:
            fully.append(sku)
        else:
            short.append(sku)
            if net <= 0 and request.priority == Priority.CRITICAL:
                missing_critical = True

    if missing_critical:
        return RoutingDecision(
            directive=Directive.DEFER_PROCUREMENT, confidence=0.95, short_skus=short
        )
    if not short:
        return RoutingDecision(
            directive=Directive.ROUTE_IMMEDIATE, confidence=1.0, short_skus=[]
        )
    return RoutingDecision(
        directive=Directive.ROUTE_PARTIAL, confidence=0.75, short_skus=short
    )

5. Compose the gate with a graceful fallback

Wire the pieces into a single async entry point that the dispatch worker awaits. When live inventory cannot be reached inside the SLA budget, degrade to a time-bound snapshot and lower the confidence rather than blocking the entire dispatch queue. Every decision is logged with a structured audit line for compliance and root-cause analysis.

import logging

logger = logging.getLogger(__name__)


async def check_parts_availability(
    client: httpx.AsyncClient,
    base_url: str,
    wo: WorkOrderPayload,
    snapshot_fallback: Dict[str, Any],
    timeout_s: float = 4.0,
) -> RoutingDecision:
    request = to_routing_request(wo)
    key = generate_idempotency_key(request.work_order_id, request.part_skus)
    try:
        inventory = await fetch_availability(
            client, base_url, request.part_skus, key, timeout_s
        )
        decision = evaluate_routing_directive(request, inventory)
    except (httpx.HTTPError, InventoryUnavailable) as exc:
        # Circuit-breaker fallback: decide on the last good snapshot, flag low trust.
        logger.warning("live inventory unreachable wo:%s err:%s", wo.work_order_id, exc)
        decision = evaluate_routing_directive(request, snapshot_fallback)
        decision = decision.model_copy(update={"confidence": decision.confidence * 0.5})
    logger.info(
        "routed wo:%s directive:%s conf:%.2f short:%s",
        wo.work_order_id, decision.directive.value, decision.confidence, decision.short_skus,
    )
    return decision

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not in the worker source. The defaults below are conservative starting points for general-purpose MRO inventory.

Validation and Testing

Allocation math is pure, so the highest-value test asserts that a fixed request and inventory state always yield the same directive. A single deterministic assertion catches accidental nondeterminism (unsorted SKUs, dict ordering, clock-dependent math) before it reaches production.

def test_critical_shortage_defers_procurement():
    request = RoutingRequest(
        work_order_id="WO-2026-0042",
        asset_id="PUMP-014",
        part_skus=["PMP-SEAL-0042", "PMP-BRG-0019"],
        required_quantities={"PMP-SEAL-0042": 2, "PMP-BRG-0019": 1},
        location_id="STORE-A",
        priority=Priority.CRITICAL,
    )
    inventory = {
        "PMP-SEAL-0042": {"on_hand": 5, "allocated": 1, "in_transit": 0},  # net 4 >= 2
        "PMP-BRG-0019": {"on_hand": 0, "allocated": 0, "in_transit": 0},   # net 0 < 1
    }
    decision = evaluate_routing_directive(request, inventory)
    assert decision.directive is Directive.DEFER_PROCUREMENT
    assert decision.short_skus == ["PMP-BRG-0019"]
    assert decision.confidence == 0.95

On a successful evaluation the gate emits a single structured line per work order — routed wo:WO-2026-0042 directive:DEFER_PROCUREMENT conf:0.95 short:['PMP-BRG-0019'] — which is the canonical signal that a decision was made against live data. A fallback decision instead emits live inventory unreachable wo:... err:... followed by a halved confidence; seeing that pair under an upstream outage confirms the circuit breaker is holding rather than indicating a fault. Assert against both lines in integration tests to verify the full request-to-decision path.

Failure Modes and Troubleshooting

Expand each scenario for the root cause, the diagnostic log excerpt, and the fix. The checklist items render as interactive checkboxes — work through them in order.

Technician dispatched to an empty bin despite a ROUTE_IMMEDIATE

• Confirm the availability endpoint returned live data, not a cached projection — a stale 200 is the usual root cause.
• Reconcile the queried stock against a physical scan through barcode & QR integration; a scan/snapshot mismatch confirms ingestion lag rather than a routing bug.
• Verify allocated is being subtracted — a reservation that the endpoint omits inflates net availability and routes against stock that is already spoken for.

Duplicate reservations after a broker redelivery

• Check for repeated routed wo:<id> lines with the same work_order_id; their presence means the idempotency key is not reaching the endpoint.
• Confirm generate_idempotency_key sorts the SKU list — an unsorted list produces a different key per delivery and defeats deduplication.
• Ensure the CMMS honors the Idempotency-Key header and returns the original decision on replay rather than re-evaluating.

The gate blocks the whole dispatch queue during an inventory outage

• Diagnostic: no routed or live inventory unreachable line for affected work orders — the call is hanging past its budget.
• Confirm AVAILABILITY_TIMEOUT_S is set and passed into fetch_availability; an unbounded request will pin a worker indefinitely.
• Verify the snapshot fallback is wired in and snapshot_max_age_s is enforced so the gate degrades instead of waiting.

Retry storms hammer the inventory API on every 4xx

• Look for tight repeated requests against a single SKU set with 400/401/404 responses.
• Confirm tenacity retries only TransportError and 5xx — a 4xx raised as InventoryUnavailable must propagate immediately.
• Check the token still carries inventory:read; an expired or under-scoped token returns 401 and should fail closed, not retry.

Frequently Asked Questions

Should the availability check be synchronous or asynchronous?

It is a synchronous decision made with asynchronous I/O. The dispatch worker awaits a single directive before it assigns a technician, but the gate fans its per-SKU queries out concurrently so latency stays inside the SLA budget regardless of how many parts a work order needs.

What happens when the inventory API is down?

The gate degrades to a time-bound snapshot and halves the confidence rather than blocking the queue. If the snapshot is older than snapshot_max_age_s it fails closed, because routing against stock that may no longer exist is worse than deferring the job.

How does this differ from automated reorder triggers?

This component only reads stock and decides whether a job can proceed; it never raises a purchase order. When it returns DEFER_PROCUREMENT, the caller forwards that to automated reorder triggers, which owns the procurement signal. Keeping the two stages separate stops routing logic from leaking into procurement.

Why hash the SKU list into the idempotency key?

So a re-sent or redelivered request is recognised as the same decision regardless of the order the SKUs arrive in. Without the sort, two identical work orders with reordered part lists would produce different keys and the endpoint could double-process them.

Related

Feed the gate accurate min/reorder levels from inventory threshold optimization, forward DEFER_PROCUREMENT decisions to automated reorder triggers, confirm physical allocation with barcode & QR integration, guarantee live reads with real-time parts availability checks via REST APIs, and align query batching with async batch processing.

Part of: Asset Lookup & Inventory Synchronization.