Add SQL table and more thorough example

Author: danielmillerp

examples/demos/procurement_agent/project/activities/activities.py

@@ -18,6 +18,11 @@
     update_delivery_date_for_item_for_workflow,
     remove_delivery_item_for_workflow,
     update_project_end_date_for_workflow,
+    create_procurement_item,
+    update_procurement_item,
+    delete_procurement_item,
+    get_procurement_item_by_name,
+    get_all_procurement_items,
     DatabaseError,
     DataCorruptionError,
 )
@@ -309,4 +314,252 @@ async def update_project_end_date(workflow_id: str, new_end_date: str) -> str:
     except Exception as e:
         # Unexpected error - log and let Temporal retry
         logger.error(f"Unexpected error updating project end date: {e}")
+        raise
+
+
+@activity.defn
+async def create_procurement_item_activity(
+    workflow_id: str,
+    item: str,
+    status: str,
+    eta: str | None = None,
+    date_arrived: str | None = None,
+    purchase_order_id: str | None = None
+) -> str:
+    """
+    Creates a new procurement item for tracking through the workflow.
+
+    Call this when:
+    - A submittal is approved (Submittal_Approved event) - automatically after submittal approval
+    - Human feedback requests creating a new procurement item
+
+    Args:
+        workflow_id: The Temporal workflow ID
+        item: The item name (e.g., "Steel Beams")
+        status: Current status of the item (e.g., "submittal_approved")
+        eta: Optional estimated time of arrival
+        date_arrived: Optional date the item arrived
+        purchase_order_id: Optional purchase order ID
+
+    Raises:
+        ApplicationError: Non-retryable if data is invalid
+        DatabaseError: Retryable if database connection fails
+    """
+    logger.info(f"Creating procurement item for workflow {workflow_id}: {item} with status {status}")
+
+    try:
+        await create_procurement_item(
+            workflow_id=workflow_id,
+            item=item,
+            status=status,
+            eta=eta,
+            date_arrived=date_arrived,
+            purchase_order_id=purchase_order_id
+        )
+        return f"Procurement item created: {item} with status {status}"
+
+    except DataCorruptionError as e:
+        # Application error - invalid data, don't retry
+        logger.error(f"Data corruption error creating procurement item: {e}")
+        raise ApplicationError(
+            f"Invalid data creating procurement item: {e}",
+            type="DataCorruptionError",
+            non_retryable=True
+        ) from e
+
+    except DatabaseError as e:
+        # Platform error - database connection issue, let Temporal retry
+        logger.warning(f"Database error creating procurement item (will retry): {e}")
+        raise  # Let Temporal retry with activity retry policy
+
+    except Exception as e:
+        # Unexpected error - log and let Temporal retry
+        logger.error(f"Unexpected error creating procurement item: {e}")
+        raise
+
+
+@activity.defn
+async def update_procurement_item_activity(
+    workflow_id: str,
+    status: str | None = None,
+    eta: str | None = None,
+    date_arrived: str | None = None,
+    purchase_order_id: str | None = None
+) -> str:
+    """
+    Updates a procurement item's fields.
+
+    Call this when:
+    - Any event occurs that changes the item's status (e.g., shipment departed, arrived, inspection scheduled/failed/passed)
+    - Human feedback requests updating the procurement item
+    - Purchase order is issued
+    - ETA is updated
+    - Item arrives at site
+
+    Args:
+        workflow_id: The Temporal workflow ID
+        status: Optional new status
+        eta: Optional new estimated time of arrival
+        date_arrived: Optional new arrival date
+        purchase_order_id: Optional new purchase order ID
+
+    Raises:
+        ApplicationError: Non-retryable if workflow_id invalid or item not found
+        DatabaseError: Retryable if database connection fails
+    """
+    logger.info(f"Updating procurement item for workflow {workflow_id}")
+
+    try:
+        await update_procurement_item(
+            workflow_id=workflow_id,
+            status=status,
+            eta=eta,
+            date_arrived=date_arrived,
+            purchase_order_id=purchase_order_id
+        )
+        return f"Procurement item updated for workflow {workflow_id}"
+
+    except DataCorruptionError as e:
+        # Application error - item not found or invalid data, don't retry
+        logger.error(f"Data corruption error updating procurement item: {e}")
+        raise ApplicationError(
+            f"Failed to update procurement item: {e}",
+            type="DataCorruptionError",
+            non_retryable=True
+        ) from e
+
+    except DatabaseError as e:
+        # Platform error - database connection issue, let Temporal retry
+        logger.warning(f"Database error updating procurement item (will retry): {e}")
+        raise  # Let Temporal retry with activity retry policy
+
+    except Exception as e:
+        # Unexpected error - log and let Temporal retry
+        logger.error(f"Unexpected error updating procurement item: {e}")
+        raise
+
+
+@activity.defn
+async def delete_procurement_item_activity(workflow_id: str) -> str:
+    """
+    Deletes a procurement item from the database.
+
+    Call this when:
+    - Human feedback explicitly requests removing/deleting an item (e.g., "remove the steel beams")
+    - Item is no longer needed in the project
+
+    Args:
+        workflow_id: The Temporal workflow ID
+
+    Raises:
+        ApplicationError: Non-retryable if workflow_id invalid or item not found
+        DatabaseError: Retryable if database connection fails
+    """
+    logger.info(f"Deleting procurement item for workflow {workflow_id}")
+
+    try:
+        await delete_procurement_item(workflow_id)
+        return f"Procurement item deleted for workflow {workflow_id}"
+
+    except DataCorruptionError as e:
+        # Application error - item not found, don't retry
+        logger.error(f"Data corruption error deleting procurement item: {e}")
+        raise ApplicationError(
+            f"Failed to delete procurement item: {e}",
+            type="DataCorruptionError",
+            non_retryable=True
+        ) from e
+
+    except DatabaseError as e:
+        # Platform error - database connection issue, let Temporal retry
+        logger.warning(f"Database error deleting procurement item (will retry): {e}")
+        raise  # Let Temporal retry with activity retry policy
+
+    except Exception as e:
+        # Unexpected error - log and let Temporal retry
+        logger.error(f"Unexpected error deleting procurement item: {e}")
+        raise
+
+
+@activity.defn
+async def get_procurement_item_by_name_activity(workflow_id: str, item: str) -> str:
+    """
+    Retrieves a procurement item by workflow ID and item name.
+
+    Call this when:
+    - You need to check the status of a specific item
+    - You need context about an item before making decisions
+    - Human feedback requests information about a specific item
+
+    Args:
+        workflow_id: The Temporal workflow ID
+        item: The item name (e.g., "Steel Beams")
+
+    Returns:
+        JSON string of the procurement item or message if not found
+
+    Raises:
+        ApplicationError: Non-retryable if input data is invalid
+        DatabaseError: Retryable if database connection fails
+    """
+    logger.info(f"Getting procurement item for workflow {workflow_id}: {item}")
+
+    try:
+        result = await get_procurement_item_by_name(workflow_id, item)
+
+        if result is None:
+            return f"No procurement item found for workflow {workflow_id} with item name: {item}"
+
+        return json.dumps(result)
+
+    except DataCorruptionError as e:
+        # Application error - invalid input, don't retry
+        logger.error(f"Data corruption error getting procurement item: {e}")
+        raise ApplicationError(
+            f"Invalid input getting procurement item: {e}",
+            type="DataCorruptionError",
+            non_retryable=True
+        ) from e
+
+    except DatabaseError as e:
+        # Platform error - database connection issue, let Temporal retry
+        logger.warning(f"Database error getting procurement item (will retry): {e}")
+        raise  # Let Temporal retry with activity retry policy
+
+    except Exception as e:
+        # Unexpected error - log and let Temporal retry
+        logger.error(f"Unexpected error getting procurement item: {e}")
+        raise
+
+
+@activity.defn
+async def get_all_procurement_items_activity() -> str:
+    """
+    Retrieves all procurement items from the database.
+
+    Call this when:
+    - You need an overview of all procurement items
+    - You need to check the status of multiple items
+    - Human feedback requests a summary of all items
+
+    Returns:
+        JSON string of all procurement items
+
+    Raises:
+        DatabaseError: Retryable if database connection fails
+    """
+    logger.info("Getting all procurement items")
+
+    try:
+        results = await get_all_procurement_items()
+        return json.dumps(results)
+
+    except DatabaseError as e:
+        # Platform error - database connection issue, let Temporal retry
+        logger.warning(f"Database error getting all procurement items (will retry): {e}")
+        raise  # Let Temporal retry with activity retry policy
+
+    except Exception as e:
+        # Unexpected error - log and let Temporal retry
+        logger.error(f"Unexpected error getting all procurement items: {e}")
         raise