Skip to main content

Loan Application - Temporal.io based Worker

This tutorial in a nutshell​

In this tutorial, we will build a simple Temporal-based Worker in Python language that will manage a Loan Application business process in KuFlow. Our example will model a Workflow whose purpose is to fill up an application form for approval. We will include some special rules, such as automatic approval for amounts below a certain quantity, and the use of different currencies which require REST calls to a third-party application.

Prerequisites​

Before starting our Worker for the first time, we must register it in KuFlow (app.kuflow.com). After familiarizing yourself with the user interface, you are ready to perform the necessary configurations for our Worker. To do so, click on the Settings menu. If you do not see this menu, you probably do not have the necessary permissions to define processes, so you will have to request it from your Organization administrators, or simply create a FREE new account.

Create the credentials for the Worker​

We will configure an APPLICATION that will provide us with the necessary credentials so that our Worker (written in Python and located in your own machine) can interface with KuFlow.

Go to the Settings > Applications menu and click on Add application. We set the name we want and save. Next, you will get the first data needed to configure our Worker.

  • Identifier: Unique identifier of the application. For this tutorial: LoanApp
    • Later in this tutorial, we will configure it in the kuflow.api.client-id property of our example.
  • Token: Password for the application.
    • Later in this tutorial, we will configure it in the kuflow.api.client-secret property of our example.

Finally, you get something like:

Create the process definition​

We need to create the definition of the process that will execute our Workflow. In this section, we will configure the KuFlow tasks of which it is made up as well as the information necessary to complete said tasks, the process access rules (i.e. RBAC), as well as another series of information. To do this we go to the Setting > Processes menu and create a new process.

A Process Definition with the following data:

  • Description
    • Free text description about the Workflow.
  • Workflow
    • Workflow Engine
      • KuFlow Engine, because we are designing a Temporal-based Worker.
    • Workflow Application
      • LoanApp, the application to which our Worker will connect to
    • Task queue
      • The name of the Temporal queue where the KuFlow tasks will be set. You can choose any name, later you will set this same name in the appropriate configuration in your Worker. For this tutorial: SampleQueue.
    • Type
      • It must match the name of the TypeScript function of the Workflow. For this tutorial, SampleLoanWorkflow is the name you should type in this input.
  • Metadata
    • Reason metadata: To require the user to indicate a personalized metadata (reason) when starting the process.
      • Code: REASON
      • Name: Reason
      • Field Type: Text
      • Mandatory: Yes
      • Editable: Yes (To allow editing until the process has been completed)
    • Reason metadata: To require the user to indicate a personalized metadata (reason) when starting the process.
      • Code: FIRST_NAME
      • Name: First name
      • Field Type: Text
      • Mandatory: No
      • Editable: No
    • Reason metadata: To require the user to indicate a personalized metadata (reason) when starting the process.
      • Code: LAST_NAME
      • Name: Last name
      • Field Type: Text
      • Mandatory: No
      • Editable: No

We must also define the Permissions

  • At least, one user or group of users must have the role of INITIATOR in order to instantiate the process through the application. In this tutorial, we will allow all users from this organization to be able to process initiation (Groups > Users)
  • At least, one user or group of users will assume the task of loans approvals. In this tutorial, we have defined the user Murph, so we will also give her the role of PROCESS MANAGER.

We define two Structure Definitions in the process with the following data:

  • Structure Data Loan Application Form:
    • Code: SD_LOAN_APPLICATION
    • Schema: 
      {
        "title": "Form",
        "type": "object",
        "properties": {
          "FIRST_NAME": {
            "type": "string"
          },
          "LAST_NAME": {
            "type": "string"
          },
          "AMOUNT": {
            "type": "number",
            "minimum": 0
          },
          "CURRENCY": {
            "type": "string",
            "enum": [
              "EUR",
              "USD",
              "GBP"
            ]
          }
        },
        "required": [
          "FIRST_NAME",
          "LAST_NAME",
          "AMOUNT",
          "CURRENCY"
        ]
      }
  • Structure Data Approve Loan Form:
    • Code: SD_APPROVE_LOAN
    • Schema: 
      {
        "title": "Form",
        "type": "object",
        "properties": {
          "FIRST_NAME": {
            "type": "string"
          },
          "LAST_NAME": {
            "type": "string"
          },
          "AMOUNT": {
            "type": "number",
            "minimum": 0
          },
          "CURRENCY": {
            "type": "string",
            "enum": [
              "EUR",
              "USD",
              "GBP"
            ]
          }
        },
        "required": [
          "FIRST_NAME",
          "LAST_NAME",
          "AMOUNT",
          "CURRENCY"
        ]
      }

We define two Tasks Definitions in the process with the following data:

  • Task "Loan Application Form"
    • Description: Free text description about the Task.
    • Code: LOAN_APPLICATION
    • Permissions:
      • Groups > Users as Candidate (in this tutorial, we will allow all users from this organization to fill up a loan application form)
    • Structure Data: Loan Application Form
  • Task "Approve Loan"
    • Description: Free text description about the Task.
    • Code: APPROVE_LOAN
    • Permissions:
      • Murph as Candidate
    • Structure Data: Approve Loan Form

We must also define the Permissions

  • At least, one user or group of users must have the role of INITIATOR in order to instantiate the process through the application. In this tutorial, we will allow all users from this organization to be able to process initiation (Default Group)
  • At least, one user or group of users will assume the task of loans approvals. In this tutorial, we have defined the user Murph, so we will also give her the role of PROCESS MANAGER.

Source code for our Worker​

The source code for this example can be obtained from our python examples repository.

However, an alternative methodology could be to download the example template code for the process we are going to implement and compare it with the complete example in our repository. To download the initial template you must access the workflow definition in the App, and once configured, use the download template option.

Main technologies used in the example​

This code will serve as a starting point for implementing our Loan Application worker. The requirements for its use are the following:

  • Poetry (>=1.3.2)
    • For python packaging and dependency management
  • Python (>=3.8)
  • IDE
    • An IDE with good Python support is necessary to work comfortably. You can use VS Code, IntelliJ Idea, Eclipse or any other with corresponding Python plugins.

To make things simpler, the following dependencies have been used in our example:

  • KuFlow Python SDK
    • Provide some activities and utilities to work with KuFlow.
  • Temporal Python SDK
    • To perform GRPC communications with the KuFlow temporal service.

Implementation details​

The workflow​

In this section, we will make the fundamental steps to creating a workflow for this business process:

  1. Users in an organization can create a Loan Application.
  2. If the loan is over 5,000 euros, manager approval is required.
  3. The user can indicate the amount requested in several currencies.
  4. A third party API will proceed to convert the currency to Euros at the current time.
  5. The manager is the only authorized user to approve loans.
  6. The user is informed of the final decision on the loan.

We modify the method that creates the Loan Form Application task so that it returns the task to the workflow upon completion.

workflow.py

async def _create_process_item_loan_application(self, process_id: str):
    """Create process item "Loan Application" in KuFlow and wait for its completion"""

    process_item_id = str(uuid7())

    create_request = models_activity.ProcessItemCreateRequest(
        id=process_item_id,
        process_id=process_id,
        type=models_rest.ProcessItemType.TASK,
        task=models_rest.ProcessItemTaskCreateParams(
            task_definition_code=SampleWorkflow._TASK_CODE_LOAN_APPLICATION_FORM
        ),
    )

    await self._create_process_item_and_wait_completion(create_request)

    retrieve_response: models_activity.ProcessItemRetrieveResponse = await workflow.execute_activity(
        KuFlowActivities.retrieve_process_item,
        models_activity.ProcessItemRetrieveRequest(process_item_id=process_item_id),
        start_to_close_timeout=SampleWorkflow._KUFLOW_ACTIVITY_START_TO_CLOSE_TIMEOUT,
        schedule_to_close_timeout=SampleWorkflow._KUFLOW_ACTIVITY_SCHEDULE_TO_CLOSE_TIMEOUT,
        retry_policy=SampleWorkflow._KUFLOW_ACTIVITY_RETRY_POLICY,
    )

    return retrieve_response.process_item


async def _create_process_item_and_wait_completion(self, request: models_activity.ProcessItemCreateRequest) -> None:
    await workflow.execute_activity(
        KuFlowActivities.create_process_item,
        request,
        start_to_close_timeout=SampleWorkflow._KUFLOW_ACTIVITY_START_TO_CLOSE_TIMEOUT,
        schedule_to_close_timeout=SampleWorkflow._KUFLOW_ACTIVITY_SCHEDULE_TO_CLOSE_TIMEOUT,
        retry_policy=SampleWorkflow._KUFLOW_ACTIVITY_RETRY_POLICY,
    )

    await workflow.wait_condition(lambda: request.id in self._kuflow_completed_task_ids)

We modify the function that manages the Loan Approval task, so that in its creation the First Name, Last Name fields are completed with the data of the initial task of the form. In the same way, the value for the Amount requested is added, once transformed to Euros from the source currency.

workflow.py

async def _create_process_item_approve_loan(
    self, process_item_loan_application: models_rest.ProcessItem, amount_eur: str
) -> models_rest.ProcessItem:
    """Create process item "Approve Loan" in KuFlow and wait for its completion"""

    # FirstName and LastName is mandatory
    first_name = str(process_item_loan_application.task.data.value.get("FIRST_NAME"))
    last_name = str(process_item_loan_application.task.data.value.get("LAST_NAME"))

    process_item_id = str(uuid7())

    create_request = models_activity.ProcessItemCreateRequest(
        id=process_item_id,
        process_id=process_item_loan_application.process_id,
        type=models_rest.ProcessItemType.TASK,
        task=models_rest.ProcessItemTaskCreateParams(
            task_definition_code=SampleWorkflow._TASK_CODE_APPROVE_LOAN,
            data=models_rest.JsonValue(
                value={"FIRST_NAME": first_name, "LAST_NAME": last_name, "AMOUNT": amount_eur}
            ),
        ),
    )

    await self._create_process_item_and_wait_completion(create_request)

    retrieve_request = models_activity.ProcessItemRetrieveRequest(
        process_item_id=process_item_id,
    )
    retrieve_response: models_activity.ProcessItemRetrieveResponse = await workflow.execute_activity(
        KuFlowActivities.retrieve_process_item,
        retrieve_request,
        start_to_close_timeout=SampleWorkflow._KUFLOW_ACTIVITY_START_TO_CLOSE_TIMEOUT,
        schedule_to_close_timeout=SampleWorkflow._KUFLOW_ACTIVITY_SCHEDULE_TO_CLOSE_TIMEOUT,
        retry_policy=SampleWorkflow._KUFLOW_ACTIVITY_RETRY_POLICY,
    )

    return retrieve_response.process_item

The notification tasks (approval or rejection) will be left as they are in the downloaded template.

We need to modify the main method of the workflow with the following content:

workflow.py

@workflow.run
async def run(self, request: models_workflow.WorkflowRequest) -> models_workflow.WorkflowResponse:
    workflow.logger.info(f"Process {request.process_id} started")

    process_item_loan_application = await self._create_process_item_loan_application(request.process_id)

    await self._update_process_metadata(process_item_loan_application)

    currency = str(process_item_loan_application.task.data.value.get("CURRENCY"))
    amount = str(process_item_loan_application.task.data.value.get("AMOUNT"))

    # Convert to euros
    amount_eur = await self._convert_to_euros(currency, amount)

    loan_authorized = True
    if float(amount_eur) > 5000:
        process_item_approve_loan = await self._create_process_item_approve_loan(
            process_item_loan_application, amount_eur
        )

        # Approval is mandatory and not multiple
        approval = str(process_item_approve_loan.task.data.value.get("APPROVAL"))
        loan_authorized = approval == "YES"

    if loan_authorized:
        await self._create_process_item_notification_of_loan_granted(request.process_id)
    else:
        await self._create_process_item_notification_of_loan_rejection(request.process_id)

    return models_workflow.WorkflowResponse(f"Completed process {request.process_id}")

The activities​

Our workflow interacts with two external systems, the first one is the KuFlow REST API. For this we use a set of activities already provided in the SDK. On the other hand, we need to connect to a third party API that provides real-time currency conversion. For this we create the following activity.

activities.py

from dataclasses import dataclass

import requests
from temporalio import activity

CONVERT_ENDPOINT = (
    "https://cdn.jsdelivr.net/gh/fawazahmed0/currency-api@1/latest/currencies"
)


@dataclass
class ConvertRequest:
    amount: float
    base_currency: str
    target_currency: str


@dataclass
class ConvertResponse:
    amount: float


class CurrencyConversionActivities:
    def __init__(self):
        self.activities = [self.convert]

    @activity.defn(name="Currency_convert")
    async def convert(self, request: ConvertRequest) -> ConvertResponse:
        # Make a GET request to the API
        response = requests.get(
            f"{CONVERT_ENDPOINT}/{request.base_currency}/{request.target_currency}.json"
        )

        # Parse the response JSON
        data = response.json()

        # Get the exchange rate
        exchange_rate = data[request.target_currency]

        # Convert
        result = exchange_rate * request.amount

        return ConvertResponse(amount=result)

The worker​

We need to indicate that our Temporal worker, in addition to implementing the KuFlow activities, implements the activity that connects to a third-party API and obtains a currency transformation.

worker.py

# Initializing KuFlow Temporal.io activities
kuflow_activities = KuFlowActivities(kuflow_rest_client)

# Initializing custom activities
currency_conversion_activities = CurrencyConversionActivities()

# Activities for the worker
activities = (
    kuflow_activities.activities + currency_conversion_activities.activities
)

# KuFlow Temporal connection
kuflow_temporal_connection = KuFlowTemporalConnection(
    kuflow=KuFlowConfig(rest_client=kuflow_rest_client),
    temporal=TemporalConfig(
        client=TemporalClientConfig(
            target_host=configuration.temporal_host,
        ),
        worker=TemporalWorkerConfig(
            task_queue=configuration.temporal_queue,
            workflows=[SampleWorkflow],
            activities=activities,
            debug_mode=True,
        ),
    ),
)

# Start temporal worker
await kuflow_temporal_connection.run_worker()

Worker configuration​

The last step of this section is filling up the application configuration information. You must complete all the settings and replace the example values. The appropriate values can be obtained from the KuFlow application. Check out the Create the credentials for the Worker section of this tutorial.

# ===================================================================
# PLEASE COMPLETE ALL CONFIGURATIONS BEFORE STARTING THE WORKER
# ===================================================================
kuflow:
  api:
    # ID of the APPLICATION configured in KUFLOW.
    # Get it in "Application details" in the Kuflow APP.
    client-id: "15a113fa-4e95-46f1-ae83-97f73be98a22"

    # TOKEN of the APPLICATION configured in KUFLOW.
    # Get it in "Application details" in the Kuflow APP.
    client-secret: FILL_ME

temporal:
  # Temporal Queue. Configure it in the "Process definition" in the KUFLOW APP.
  kuflow-queue: "SampleQueue"

Please note that this is a YAML, respect the indentation.

We can test this scenario, initiating the process several times, in order to create different applications and approvals

Summary​

In this tutorial, we have covered the basics of creating a workflow in KuFlow for loan applications. We have defined a new process definition and we have built a workflow that contemplates the required business rules defined above.

We sincerely hope that this step-by-step guide will help you to understand better how KuFlow can help your business to have better and more solid business processes.

Kuflow Logo
Legal
Copyright © 2024 KUFLOW, S.L.