NOTICE: All information contained herein is, and remains
the property of TechnoCore.
The intellectual and technical concepts contained
herein are proprietary to TechnoCore and dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from TechnoCore.
The ObjWorkflowService module implements the SERVICE node type for workflows. SERVICE nodes dynamically load and execute custom service classes that contain business logic. This provides a flexible mechanism for extending workflow functionality without modifying the core workflow engine.
SERVICE nodes enable:
ObjWorkflowNode (base class)
└── ObjWorkflowService
In def_workflows table:
INSERT INTO def_workflows (WorkflowName, Package, Rank, Type, Name, BranchDirect, Async)
VALUES ('USER_REGISTRATION', 'homechoice', 20, 'SERVICE', 'ValidateUser', 'NEXT_NODE', 'N');
The node Name field is looked up in def_service to find the actual service class:
-- Service mapping
INSERT INTO def_service (ServiceCode, ServiceName, Package)
VALUES ('ValidateUser', 'ObjServiceUserValidation', 'homechoice');
All service classes must implement:
class ObjServiceExample:
def __init__(self, DB):
self.DB = DB
self.Param1 = ""
self.Param2 = ""
self.Param3 = ""
# ... other attributes ...
def run_workflow_direct(self, guid: str, selector: str) -> str:
"""
Execute service logic.
Args:
guid: Input GUID for data
selector: Service selector/mode
Returns:
Result string or dict
"""
# Implement service logic here
return "SUCCESS"
The SERVICE node automatically maps context values to service object attributes:
# run_context['data'] → service._calc_*
run_context['data'] = {'total': '1000', 'count': '5'}
# Results in:
# service._calc_total = "1000"
# service._calc_count = "5"
# run_context['result'] → service.*
run_context['result'] = {'_form_username': 'john', '_api_status': 'ok'}
# Results in:
# service._form_username = "john"
# service._api_status = "ok"
# Param1, Param2, Param3 from context
run_context['param1'] = 'USER123'
# Results in:
# service.Param1 = "USER123"
# service.Param2 = "" (default)
# service.Param3 = "" (default)
Services can receive JSON configuration via the Node_Data column:
UPDATE def_workflows
SET Node_Data = '{"threshold": 100, "mode": "strict"}'
WHERE WorkflowName = 'USER_REGISTRATION'
AND Name = 'ValidateUser';
Accessed in service:
def run_workflow_direct(self, guid, selector):
threshold = self.node_data.get('threshold', 50)
mode = self.node_data.get('mode', 'normal')
# ...
When Async = 'Y', the service is queued for background processing:
{
"_id": "service_uuid_12345",
"ServiceCode": "ValidateUser",
"Guid": "service_uuid_12345",
"SourceGuid": "USER123",
"Package": "HOMECHOICE",
"Status": "pending", # Changes to "processing" → "complete"
"context": { ... }, # Full run_context
"createdAt": ISODate("2025-12-14T10:00:00Z")
}
{package}_queue_service
Example: homechoice_queue_service
A separate process monitors the queue:
# Pseudo-code for worker
while True:
pending = mongo.find({"Status": "pending"})
for task in pending:
service_obj.run_workflow_direct(task.Guid, task.ServiceCode)
mongo.update({"_id": task._id}, {"Status": "complete"})
def run_workflow_direct(self, guid, selector):
return "SUCCESS" # Stored in run_context['result']['_service_validateuser']
def run_workflow_direct(self, guid, selector):
return {
"Status": "APPROVED",
"Score": 85,
"Message": "User validated"
}
# Results in run_context['result']['_service_validateuser'] = {
# "status": "APPROVED",
# "score": 85,
# "message": "User validated"
# }
Result: "FAIL: Missing Service {name}"
Result: "FAIL: Missing method run_workflow_direct"
# Malformed JSON in Node_Data
run_context['service_validateuser'] = "FAILED"
# Workflow continues with this result
# factory.service/package.homechoice/ObjServiceUserValidation.py
import os
import sys
base_path = os.getcwd()
sys.path.append(base_path + "/factory.core")
from ObjData import ObjData
class ObjServiceUserValidation(ObjData):
def __init__(self, DB):
super().__init__(DB)
self.Param1 = ""
self.Param2 = ""
self.Param3 = ""
self.node_data = {}
def run_workflow_direct(self, guid: str, selector: str) -> dict:
"""Validate user based on criteria"""
# Access context attributes set by SERVICE node
username = getattr(self, '_form_username', '')
email = getattr(self, '_form_email', '')
# Access node data configuration
strict_mode = self.node_data.get('mode') == 'strict'
# Validation logic
if not username or not email:
return {
"Status": "INVALID",
"Message": "Missing required fields"
}
# Database check
sql = f"SELECT COUNT(*) FROM users WHERE username = '{username}'"
exists = int(self.sql_get_value(sql) or 0)
if exists > 0:
return {
"Status": "DUPLICATE",
"Message": "Username already exists"
}
return {
"Status": "VALID",
"Message": "User validation passed"
}
-- Main workflow
INSERT INTO def_workflow (WorkflowName, Package, Description, Active)
VALUES ('USER_REGISTRATION', 'homechoice', 'User registration workflow', 'Y');
-- Service definition
INSERT INTO def_service (ServiceCode, ServiceName, Package, Description)
VALUES ('ValidateUser', 'ObjServiceUserValidation', 'homechoice', 'Validate user registration');
-- Workflow step using SERVICE node
INSERT INTO def_workflows
(WorkflowName, Package, Rank, Type, Name, BranchDirect, Async, Node_Data)
VALUES
('USER_REGISTRATION', 'homechoice', 20, 'SERVICE', 'ValidateUser', 'CHECK_RESULT', 'N',
'{"mode": "strict", "check_duplicates": true}');
-- Conditional branching based on service result
INSERT INTO def_workflows
(WorkflowName, Package, Rank, Type, Name, BranchSql)
VALUES
('USER_REGISTRATION', 'homechoice', 30, 'DECISION', 'CHECK_RESULT',
"SELECT CASE WHEN '$_service_validateuser$' = 'VALID' THEN 'APPROVE' ELSE 'REJECT' END");
ObjServiceUserValidation not ObjServiceUVObjService for consistencypackage.{name} foldersself.debug() or self.error()Version: 8.0
Last Updated: 2025-12-14