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
│ ├─ ConfigIni.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: