mirror/temporal-ai-agent
1
0
Fork 0
mirror of https://github.com/temporal-community/temporal-ai-agent.git synced 2026-03-15 14:08:08 +01:00
Code Issues Packages Projects Releases Wiki Activity

well it kinda works

Browse Source
This commit is contained in:
Joshua Smith
2025-04-12 15:41:47 -04:00
parent f0524f2b5f
commit 79dcd40dde
3 changed files with 47 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
tools/fin/submit_loan_application.py
View File

@@ -1,8 +1,14 @@
from datetime import timedelta
from datetime import date, timedelta
import os
from pathlib import Path
import json
from temporalio.client import Client, WorkflowHandle
from temporalio.client import (
Client,
WithStartWorkflowOperation,
WorkflowHandle,
WorkflowUpdateFailedError,
)
from temporalio import common
from dataclasses import dataclass
from typing import Optional
import asyncio
@@ -15,12 +21,13 @@ from shared.config import get_temporal_client
@dataclass
class TransactionRequest:
amount: float
account_id: str
sourceAccount: str
targetAccount: str
@dataclass
class TxResult:
transaction_id: str
message: str
transactionId: str
status: str
#demonstrate starting a workflow and early return pattern while the workflow continues
async def submit_loan_application(args: dict) -> dict:
@@ -29,7 +36,11 @@ async def submit_loan_application(args: dict) -> dict:
loan_status: dict = await start_workflow(amount=amount,account_name=account_key)
return {'status': loan_status.get("loan_status"), 'detailed_status': loan_status.get("results"), 'next_step': loan_status.get("advisement"), 'confirmation_id': loan_status.get("workflowID")}
if loan_status.get("error") is None:
return {'status': loan_status.get("loan_application_status"), 'detailed_status': loan_status.get("application_details"), 'next_step': loan_status.get("advisement"), 'confirmation_id': loan_status.get("transaction_id")}
else:
print(loan_status)
return loan_status
# Async function to start workflow
@@ -40,55 +51,53 @@ async def start_workflow(amount: str, account_name: str, )-> dict:
start_real_workflow = os.getenv("FIN_START_REAL_WORKFLOW")
if start_real_workflow is not None and start_real_workflow.lower() == "false":
START_REAL_WORKFLOW = False
return {'loan_application_status': "applied", 'application_details': "loan application is submitted and initial validation is complete",'transaction_id': "APPLICATION"+account_name, 'advisement': "You'll receive a confirmation for final approval in three business days", }
else:
START_REAL_WORKFLOW = True
if START_REAL_WORKFLOW:
# Define the workflow ID and task queue
workflow_id = "APPLICATION-"+account_name
workflow_id = "LOAN_APPLICATION-"+account_name+"-"+date.today().strftime('%Y-%m-%d')
task_queue = "LatencyOptimizationTEST"
# Create a TransactionRequest (matching the Java workflow's expected input)
tx_request = TransactionRequest(
amount=float(amount),
account_id=account_name
targetAccount=account_name,
sourceAccount=account_name,
)
#try:
# Use update-with-start to start the workflow and call the update method
handle: WorkflowHandle = await client.start_workflow(
"TransactionWorkflowLocalBeforeUpdate.processTransaction", # Workflow name
tx_request, # Input to the processTransaction method
start_op = WithStartWorkflowOperation(
"TransactionWorkflowLocalBeforeUpdate",
tx_request,
id=workflow_id,
id_conflict_policy=common.WorkflowIDConflictPolicy.USE_EXISTING,
task_queue=task_queue,
execution_timeout=timedelta(minutes=5),
# Specify the update to call immediately after starting
update="returnInitResult",
update_args=[] # No arguments needed for returnInitResult
)
# Wait for the update result (returnInitResult)
update_result = await handle.result_of_update("returnInitResult")
# Since the result is a RawValue, we need to deserialize it
# For simplicity, assuming the result is TxResult (adjust based on actual serialization)
#result_dict = update_result.payloads[0].decode() # Simplified; use proper deserialization
tx_result = TxResult(
transaction_id=result_dict.get("transaction_id", ""),
message=result_dict.get("message", "")
)
try:
print("trying update-with-start")
tx_result = TxResult(
await client.execute_update_with_start_workflow(
"returnInitResult",
start_workflow_operation=start_op,
)
)
except WorkflowUpdateFailedError:
print("aww man got exception WorkflowUpdateFailedError" )
tx_result = None
return_msg = "Loan could not be processed for " + account_name
return {"error": return_msg}
print(f"Update result: Transaction ID = {tx_result.transaction_id}, Message = {tx_result.message}")
workflow_handle = await start_op.workflow_handle()
print(tx_result)
print(f"Update result: Transaction ID = {tx_result.transactionId}, Message = {tx_result.status}")
# Optionally, wait for the workflow to complete and get the final result
# final_result = await handle.result()
# print(f"Workflow completed with result: {final_result}")
#except Exception as e:
# print(f"Error executing workflow: {e}")
# return {'status': loan_status.get("loan_status"), 'detailed_status': loan_status.get("results"), 'next_step': loan_status.get("advisement"), 'confirmation_id': loan_status.get("workflowID")}
return {'status': "status", 'detailed_status': "loan application is submitted and initial validation is complete",'confirmation id': "11358", 'next_step': "You'll receive a confirmation for final approval in three business days", }
# return {'status': loan_status.get("loan_status"), 'detailed_status': loan_status.get("results"), 'next_step': loan_status.get("advisement"), 'confirmation_id': loan_status.get("workflowID")}
return {'loan_application_status': "applied", 'application_details': "loan application is submitted and initial validation is complete",'transaction_id': tx_result.transactionId, 'advisement': "You'll receive a confirmation for final approval in three business days", }