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.
Version: 8.0
Module: factory.core/ObjTypes.py
ObjTypes provides a centralized type alias system for the Axion framework, improving code readability, maintainability, and type consistency. Instead of using verbose type annotations like Dict[str, Any] throughout the codebase, we use semantic type aliases that convey meaning and intent.
ConfigDict is more meaningful than Dict[str, Any]DatabaseConnectionTypePurpose: Represents all supported database connection types plus int (0 = no connection)
Supported Connections:
pymysql)pymssql)sqlite3)pymongo)influxdb_client)psycopg2)Usage:
from ObjTypes import DatabaseConnectionType
class MyClass(ObjData):
def __init__(self, db: DatabaseConnectionType = 0):
super().__init__(db)
WorkflowContextTypePurpose: Execution context that flows through workflow nodes
Structure: dict[str, Union[str, int, dict, list, None]]
Contains:
Usage:
from ObjTypes import WorkflowContextType
def execute(self, run_context: WorkflowContextType, ...) -> NodeExecuteResultType:
result = run_context.get("result", {})
# Process workflow context
NodeExecuteResultTypePurpose: Standard return type for workflow node execute() methods
Structure: tuple[WorkflowContextType, str]
Returns: (updated_context, current_result_string)
Usage:
from ObjTypes import NodeExecuteResultType
def execute(self, ...) -> NodeExecuteResultType:
# Process node
return run_context, current_result
WorkflowResultSetPurpose: Extracted result data from workflow execution context
Structure: dict[str, Any]
Contains: Subset of WorkflowContextType containing only result fields
Usage:
from ObjTypes import WorkflowResultSet
def render_context_resultset(self, run_context: JsonDict) -> None:
result_set: WorkflowResultSet = {}
if "result" in run_context:
result_set = run_context["result"]
WorkflowNodeListPurpose: List of workflow node identifiers
Structure: list[str]
Used For:
Usage:
from ObjTypes import WorkflowNodeList
def parse_transitions(self, transitions: WorkflowNodeList) -> WorkflowNodeList:
nodes: WorkflowNodeList = []
# Process transitions
return nodes
WorkflowNodeDataListPurpose: List of node data from database queries
Structure: list[dict[str, Any]]
Contains: Structured node configuration data from def_workflows table
Usage:
from ObjTypes import WorkflowNodeDataList
nodes: WorkflowNodeDataList = self.sql_get_dictionary_list(sql)
JsonDictPurpose: Generic JSON-like dictionary structure
Structure: dict[str, Any]
Used For:
Usage:
from ObjTypes import JsonDict
def parse_bpmn(self, file_name: str) -> None:
bpmn: JsonDict = xmltodict.parse(xml_data)
if "PMML" in bpmn:
data = bpmn["PMML"]
JsonDictListPurpose: List of JSON-like dictionaries
Structure: list[dict[str, Any]]
Used For:
sql_get_dictionary_list()Usage:
from ObjTypes import JsonDictList
nodes: JsonDictList = self.sql_get_dictionary_list(sql)
for node in nodes:
print(node["name"])
StringListPurpose: List of strings
Structure: list[str]
Used For:
Usage:
from ObjTypes import StringList
def get_primary_keys(self, table: str) -> StringList:
primary_keys: StringList = []
# Get keys
return primary_keys
ConfigDictPurpose: Configuration dictionary structures from config.yaml
Structure: dict[str, Any]
Used For:
Usage:
from ObjTypes import ConfigDict
config_buffer: Optional[ConfigDict] = None
aws_secrets_cache: ConfigDict = {}
ConfigSectionPurpose: Specific section of configuration file
Structure: dict[str, Any]
Used For:
Usage:
from ObjTypes import ConfigSection
def get_database_config(self) -> ConfigSection:
return self.config_buffer["database"]
QueryResultListPurpose: Results from sql_get_list() - list of tuples or lists
Structure: list[tuple | list]
Used For:
Usage:
from ObjTypes import QueryResultList
results: QueryResultList = self.sql_get_list(sql)
for row in results:
column1, column2 = row[0], row[1]
QueryRowPurpose: Single row from database query
Structure: tuple | list
Used For:
Usage:
from ObjTypes import QueryRow
row: QueryRow = self.sql_get_values(sql)
workflow, version = row[0], row[1]
ApiPayloadPurpose: Request/response payload for API calls
Structure: dict[str, Any]
Used For:
Usage:
from ObjTypes import ApiPayload
def call_api(self, endpoint: str, payload: ApiPayload) -> str:
response = requests.post(endpoint, json=payload)
return response.text
WebhookPayloadPurpose: Webhook request payload
Structure: dict[str, Any]
Similar to: ApiPayload but semantically indicates webhook context
Usage:
from ObjTypes import WebhookPayload
def handle_webhook(self, payload: WebhookPayload) -> None:
event_type = payload.get("event_type")
# Process webhook
CalculationContextPurpose: Execution context for calculation services
Structure: dict[str, Any]
Contains:
Usage:
from ObjTypes import CalculationContext
def calculate(self, context: CalculationContext) -> None:
params = context.get("parameters", {})
# Perform calculation
ServiceContextPurpose: Generic execution context for service objects
Structure: dict[str, Any]
Contains:
Usage:
from ObjTypes import ServiceContext
def execute_service(self, context: ServiceContext) -> None:
service_name = context.get("service_name")
# Execute service
SimulationContextPurpose: Execution context for simulation runs
Structure: dict[str, Any]
Contains:
Usage:
from ObjTypes import SimulationContext
def run_simulation(self, context: SimulationContext) -> None:
simulation_guid = context.get("simulation_guid")
# Run simulation
| Type Alias | Purpose | Common Usage |
|---|---|---|
DatabaseConnectionType |
DB connections | __init__(db: DatabaseConnectionType = 0) |
WorkflowContextType |
Workflow execution context | execute(run_context: WorkflowContextType) |
NodeExecuteResultType |
Node return type | def execute(...) -> NodeExecuteResultType |
JsonDict |
Generic JSON data | bpmn: JsonDict = xmltodict.parse(...) |
JsonDictList |
List of JSON objects | nodes: JsonDictList = sql_get_dictionary_list(...) |
StringList |
List of strings | def get_keys() -> StringList |
ConfigDict |
Configuration data | config_buffer: Optional[ConfigDict] |
ApiPayload |
API request/response | def call_api(payload: ApiPayload) |
WebhookPayload |
Webhook data | def handle_webhook(payload: WebhookPayload) |
CalculationContext |
Calculation context | def calculate(context: CalculationContext) |
from typing import Dict, Any, List
def get_workflow_json(self, workflow_code: str) -> Optional[Dict[str, Any]]:
nodes_data: List[Dict[str, Any]] = self.sql_get_dictionary_list(sql)
result: Dict[str, Any] = {}
return result
from typing import Optional
from ObjTypes import JsonDict, JsonDictList
def get_workflow_json(self, workflow_code: str) -> Optional[JsonDict]:
nodes_data: JsonDictList = self.sql_get_dictionary_list(sql)
result: JsonDict = {}
return result
from ObjTypes import JsonDict, StringList, ...Dict[str, Any] → JsonDict (or more specific like ConfigDict, ApiPayload)List[str] → StringListList[Dict[str, Any]] → JsonDictListDict, Any, List if no longer neededpython3 -m py_compile <file> to check syntax✅ DO use type aliases when:
❌ DON'T use type aliases when:
int, str, bool)For configuration data:
ConfigDict for general config structuresConfigSection for specific config sectionsFor API/communication:
ApiPayload for REST API callsWebhookPayload for webhook handlersJsonDict for generic JSON dataFor workflow:
WorkflowContextType for node execution contextWorkflowResultSet for extracted resultsWorkflowNodeList for node identifiersFor lists:
StringList for list of stringsJsonDictList for list of dictionariesQueryResultList for raw SQL resultsCreate a new type alias when:
EmailTemplate, NotificationPayload)Example - Adding a new type:
# In ObjTypes.py
# Email & Notification Type Aliases
# EmailTemplate: Email template configuration and content
# Contains template variables, subject, body, and formatting
type EmailTemplate = dict[str, Any]
# NotificationPayload: Notification data for Discord, Slack, etc.
# Contains notification type, message, and channel information
type NotificationPayload = dict[str, Any]
Then add to __all__:
__all__ = [
# ... existing types ...
'EmailTemplate',
'NotificationPayload',
]
from ObjTypes import ConfigDict
class MyService:
config: ConfigDict = {}
def load_config(self, filename: str) -> ConfigDict:
with open(filename) as f:
self.config = yaml.safe_load(f)
return self.config
from ObjTypes import ApiPayload, JsonDict
def call_external_api(self, endpoint: str, payload: ApiPayload) -> JsonDict:
response = requests.post(endpoint, json=payload)
return response.json()
from ObjTypes import WorkflowContextType, NodeExecuteResultType
def execute(
self,
run_context: WorkflowContextType,
current_result: str,
**kwargs
) -> NodeExecuteResultType:
# Process node
return run_context, current_result
from ObjTypes import JsonDictList, StringList
def get_workflow_nodes(self, workflow_code: str) -> JsonDictList:
sql = self.get_queries("get_nodes").format(workflow=workflow_code)
nodes: JsonDictList = self.sql_get_dictionary_list(sql)
return nodes
def get_primary_keys(self, table: str) -> StringList:
sql = self.get_queries("get_primary_keys").format(table=table)
keys: StringList = self.sql_get_list(sql)
return keys
ObjTypes.py (Central Type Definitions)
↓
├─→ Workflow Modules (factory.workflow)
│ ├─ ObjWorkflow.py
│ ├─ ObjWorkflowEdit.py
│ ├─ ObjWorkflowVisual.py
│ ├─ ObjWorkflowSimul.py
│ └─ ObjWorkflowSerializer.py
│
├─→ Core Modules (factory.core)
│ ├─ ObjSimulation.py
│ ├─ ObjApi.py
│ ├─ ObjConfig.py
│ ├─ ObjFeatureStore.py
│ ├─ ObjConversation.py
│ └─ ... (16+ more files)
│
└─→ Other Factory Modules
├─ factory.service
├─ factory.webhook
├─ factory.web
└─ factory.export
Type Aliases:
EmailTemplate - Email template structuresNotificationPayload - Discord/Slack notification dataModelConfig - ML model configurationPredictionResult - ML prediction outputsMetricsData - Monitoring metricsHealthCheckResult - Health check responses__all__ export listLast Updated: 2026-02-13
Maintained By: Axion Development Team
Related Documentation: