^{_{Author – Simran Singh, Full Stack Developer}}

While making a custom flow-engine from scratch can initially appear intimidating, it’s well worth the challenges. With your own unique flow-engine, you have the ability to change its components to meet the needs of your project and to expand its capabilities to accommodate any new requirements. 

We’ll be mainly looking at automating our marketing tasks with flow-engines, even though the application of it is very vast. You will also find examples to help you create your own flow-engine. 

The need

For our marketing campaigns, every client had their own custom needs.

Some wanted to send marketing messages after a custom delay, some to send them only to a specified group of users, and some wanted to send different marketing messages based on the products that the user purchased.

When our client pool was small, we used to hard-code every client request and code the logic required for their business manually.

This was clearly not a scalable solution and it became evident as our client pool grew. We needed something that gives us the power to have configurable and complex business logic without actually having to code.

This is why we decided to build a flow system, which will help us configure business logic easily, from scratch. flow system

The structure

Typically, a flow system is represented in form of a directed graph, consisting of nodes and edges.

Imagine that each node represents a specific task, and after executing that task the next node(s) in the graph is executed, right up until the whole execution is completed.

Let’s get familiar with some terms before moving on:

• Flow: This is a graph that consists of nodes and edges which represent our business logic. This has to be a directed graph as we need to know the direction of the flow. You can also implement it to be a DAG (Directed acyclic graph) as well if you do not want loops in your business logic, but this entirely depends on your use case. For our use case, we went with a DAG.
  
• Node: Represents an action in our business logic.
  
• Flow engine: Represents the runtime of our flow system which processes and executes the flow.

Representation of the flow

Since the flow is essentially a graph, you can represent it in many ways, like an adjacency matrix, edge list, adjacency list, etc.

We will be going forward with something very simple for this example. We will assume that each node can have a maximum of one edge to the next node and thus, we can simply represent a node in this format:

{
    "$NODE_ID": {
        "type": "$NODE_TYPE",
        "props": {
            $NODE_PROPERITES..
        },
        "nextNode": "$NEXT_NODE_ID"
    },
    "$NODE_ID_2" : {
        ...
    },
    ...
}

This is a rather uncommon way of representing a graph, but it is simple to understand.

Implementation

A quality implementation of an extendable flow-engine has these 3 qualities.

• Easily extendable: We should be able to extend the system easily, i.e., add new node types.
  
• API configurable and extendable: We know how much of a pain it is to write validation logic in API for every new addition we do; we wanted something which will automatically get validation logic from node definition and didn’t require us to touch the validation logic.
  
• Easy to maintain

Implementing an Extendable Flow Engine:

Alright, now that we know what we need to build, the question is how do we build it while satisfying the requirements stated above. 

The answer is **SOLID principles of OOPs**

We will be using Python’s Django for this particular example.

To do this, you need to understand OOPS and SOLID principles first. Let’s start by defining base classes first.

Base Classes

To meet our requirement of extendable API validation logic, we will be using nested validators (called serializer in django).

First, let’s see what the base class of a node will look like:

from abc import ABC, abstractmethod
from rest_framework import serializers

class NodeConfigSerializer(serializers.Serializer):
    """Node Config serializer"""

    nextNode = serializers.CharField(null=True)
    type = serializers.CharField()
    props = serializers.JSONField()

class Node(ABC):
    """Node base class"""

    def __init__(self, context, config):
        """
        Context: A single object to access all the information regarding current context, 
            It will contain information about current context and data returned from previous node.
        Config: Node configuration
        """
        self.context = context
        self.config = config

    @abstractmethod
    def execute(self) -> Optional[Dict]:
        pass
        """Perform the task, 
        Returns data that can be passed to next node in its context if needed"""

    @abstractmethod
    def get_next_node(self):
        pass
        """Get ID of next node to execute"""

We will define a validator (serializer) along with our node base class in the same file, which will help us in increasing cohesion in our programming structure and thus lead to easier maintainability and extendability.

Let’s see how you can define a sample node to send a message, we will assume that its config requires as template_id which is to be sent in in message.

class MessageNodeSerializer(NodeConfigSerializer):
    """We inherit NodeConfigSerializer and extend it"""
    def validate_props(self, props):
        """Custom logic for validating props"""
        if "template_id" not in props:
            raise serializers.ValidationError("template_id missing")
        return props 

class MessageNode(Node):
    def execute(self):
        props = self.config['props']
        user_phone = context.user.phone 
        send_message(user_phone, props['tempplate_id']) # send message to user_phone with template_id configured in flow 
    
    def get_next_node_id(self):
        if 'nextNode' in self.config:
            return self.config['nextNode']
        return None

Factory Methods

Factory method is a design pattern in OOPs used to create the object of a specified class easily.

"""
Mapping $NODE_TYPE to Node classes
We map it to Tuple(NodeClass, NodeSerializerClass)
"""
NODES = {
    'message_node': (MessageNode, MessageNodeSerializer),
    'conditional_node': (ConditionalNode, ConditionalNodeSerializer),
    ...
}

def get_node_config_serializer(config) -> NodeConfigSerializer:
    """
    Factory method for getting node config serializer
    config: JSON config for a node
        example: 
        {
            "type": "$NODE_TYPE",
            "props": {
                $NODE_PROPERITES..
            },
            "nextNode": "$NEXT_NODE_ID"
        }
    """ 
    node_type = config.get('type')
    if node_type not in NODES:
        raise Exception("Invalid node type")
    return NODES[node_type][1](data=config)

def get_node(context, config) -> Node:
    """Factory method for getting node instance"""
    node_type = config.get('type')
    if node_type not in NODES:
        raise Exception("Invalid node type")
    return NODES[node_type][0](context, config)

Flow Engine – Tying it all together

The flow engine is the heart of our flow system, let’s see how it ties all of the things we have implemented together.

This is a very simplified format of the flow engine for easier understanding:

def execute_node(node_id, flow_graph, context):
    """
    node_id: Node ID to execute
    flow_graph: JSON representation of our graph
    context: context object to be passed to node
    """
    node_config = flow_graph[node_id]
    node = get_node(context, node_config)
    result = node.execute()
    context.previous_node_result = result 
    next_node_id = node.get_next_node_id()
    execute_node(next_node_id, flow_graph, context)

To trigger the flow initially you have to call the execute_node() function with the start_node_id which we store separately.

Extendability in API by using our factory method for validators

Now, let’s see how we tackle the problem of making API validation logic easily extendable by using our implementation strategy specified above.

We defined a validator(serializer) for each of our nodes along with Node classes, and we have also defined a factory method for getting the same. e will be using the factory method for the validator(serializer) in our API validation logic so that the API validation logic for our graph does not need to update.

Here’s the snippet we used:

# IN API validation logic: 
flow_graph = request.post_data["flow_graph"]

for node_id, node_config in flow_graph.items():
    node_config_serializer = get_node_config_serializer(config)
    if not node_config_serializer.is_valid(): # django's serializer's inbuilt method for vaildation
        raise Exception("Invalid configuration for one of the node")

Conclusion

We implemented a system that can now be termed as scalable, code scalability

This system is:

• Easily extendable: Adding a new type of node in our business logic is now as simple as inheriting the node base class and defining its validator in the same file, then mapping it in our factory method’s NODES variable.
• Maintainable: All the code related to one node, be its execution logic or validation logic is in one file, one only needs to touch that file to modify its behavior.
• Supports quicker development: Just add new classes and map them in your factory method and you are good to go.

Improvements and future scope

To continuously create a robust flow engine, we will need to optimize and build for ever-changing business requirements and scalability. Here are a few requirements we will focus on next

• Flow can support multiple edges from a node by changing flow engine logic for getting the next node(s) and how a node is represented
• Add tracking in the flow engine to track the state of flow in case there is any error.
• Change flow engine to use async methods along with some messaging queue for node’s execution for a scalable system.
• We hope this masterclass and deep dive into how we built an extendable flow engine from scratch helps you in your next project. 

Stay tuned for Part 2!

Looking for more tech deep dives? Check out How We Scaled our Django Server and Celery Workers to Handle 10X Load this post here.