ObjServiceInovo.py integrates Axion with the Inovo Lead/BulkUpload endpoint.
It supports two main operational paths:
def_service.payloadsql.removecall commands staged in environment tables:
data_api_batch_clear_command_uatdata_api_batch_clear_command_prodBoth modes post JSON payloads to Inovo and write batch outcomes to environment-specific log tables.
The service also supports a scheduled cycle mode that, per environment (UAT then PROD):
removecall entries.leadsource.data_api_batch_job_log.data_api_batch_history only after successful send.started / finished + result + duration).Module SQL is externalised in:
factory.service/package.homechoice/ObjServiceInovo.yamlObjServiceInovo.py reads these queries using get_queries() with safe inline fallbacks.
Triggered when Param1 is bulk or bulksend.
Flow:
payloadsql, endpoint, API key from def_service + def_remoteconnections.payloadsql into a DataFrame.batchid (or default to BULK_1 if not provided)./Lead/BulkUpload.data_api_batch_job_log.Triggered when Param1 is one of:
clearclearqueueclear_queuebulkclearremovecallclearcallsFlow:
data_api_batch_clear_command_uat and data_api_batch_clear_command_prod.leadsource filter.removecall records and POST in chunks.data_api_batch_job_log_uatdata_api_batch_job_log_prodTriggered when Param1 is blank or one of:
cyclescheduledschedulerrunFlow:
uat, prod) check clear staging row count.{remoteurl}/Lead/BulkUpload
apikey: <remotepassword from def_remoteconnections>Content-Type: application/jsonUser-Agent: axionTLS note:
verify=False for HTTPS requests because the target Inovo endpoint runs with internal/self-signed certificates in the current environment.{
"leadSource": "HomeChoice Collections New Cycle",
"data": [
{
"command": "removecall",
"externalid": "test456",
"customerId": "test123",
"serviceId": 321,
"loadId": 0,
"sourceId": 0,
"status": 1,
"phone": "0",
"idnumber": "0"
}
]
}
ObjServiceInovo resolves connection metadata from:
def_service (ServiceCode='INOVO_UAT' and ServiceCode='INOVO_PROD')def_remoteconnections joined via service.RemoteConnection = rm.remoteRequired fields:
payloadsql (for bulk mode)remoteurlremotepassword (used as API key)Bulk sends do not use a fixed table name in code. They use the SQL configured in:
def_service.payloadsql for INOVO_UATdef_service.payloadsql for INOVO_PRODIn practice this usually points to environment-specific bulk staging tables, for example:
data_api_batch_bulk_uatdata_api_batch_bulk_prodExpected columns from payloadsql result set:
leadsourcebatchid (optional, defaults to BULK_1 when missing)jobid (optional, defaults to env bulk job id when missing)commandidnumberexternalidserviceidloadidstatusPhonename (optional)scheduledate (optional)priority (optional)comments (optional)alternativePhone / alternativePhoneDescription (optional JSON/list text)customdata1..customdata3 and externalData1..externalData6 (optional)Expected staging tables:
data_api_batch_clear_command_uatdata_api_batch_clear_command_prodExpected columns used by service:
GuidLoadtimeleadsourcecommandidnumberexternalidserviceidloadidsourceidstatusPhoneIf Guid is blank, the row is skipped.
Target log tables:
data_api_batch_job_log_uatdata_api_batch_job_log_prodRows are written with:
jobidbatchidbatchsenttimebatchresult (includes HTTP status and response text)batchvolumeHistory tables:
data_api_batch_history_uatdata_api_batch_history_prodBehavior:
CREATE TABLE IF NOT EXISTS data_api_batch_history_uat LIKE data_api_batch_clear_command_uatCREATE TABLE IF NOT EXISTS data_api_batch_history_prod LIKE data_api_batch_clear_command_prod# Default bulk test route
python3 factory.service/package.homechoice/ObjServiceInovo.py test
# Clear mode (all staged rows)
python3 factory.service/package.homechoice/ObjServiceInovo.py clear
# Clear mode with leadsource filter
python3 factory.service/package.homechoice/ObjServiceInovo.py clear "HomeChoice Collections New Cycle"
# Clear mode with leadsource + chunk size
python3 factory.service/package.homechoice/ObjServiceInovo.py clear "HomeChoice Collections New Cycle" 200
from ObjServiceInovo import ObjServiceApi
svc = ObjServiceApi()
# Bulk mode
svc.run_workflow_direct("Bulk")
# Clear mode
svc.run_workflow_direct("CLEAR", "clear", "HomeChoice Collections New Cycle", "200")
Connect() derives mode from Param1 (or Guid fallback).
send_bulk_clear(...).bulk/bulksend route to send_bulk().cycle/scheduled/scheduler/run (and default) route to run_scheduled_cycle(...).Connect() publishes an MQTT heartbeat payload.data_api_batch_history.data_api_batch_job_log.customerId): falls back to customerId/customerid/externalid/idnumber.data_api_batch_history before removal from staging.LOCK TABLES ... WRITE / UNLOCK TABLES) so insert/verify/delete executes atomically per worker.Connect() acquires a singleton lock before running so only one Inovo service execution can run at a time; parallel runs return a busy message.started and finished events with result and durationSeconds.Common return messages:
Missing staging tables: data_api_batch_clear_command_uat, data_api_batch_clear_command_prodNo INOVO endpoint configured.No INOVO API key configured.No staged clear commands found.No payload SQL configured for INOVO service.No records to send.Unit tests for this service are in:
resource.test/pytests/factory.service/package.homechoice/test_ObjServiceInovo.pyCovered areas include:
Command executed:
dev-env/bin/python -m pytest -q resource.test/pytests/factory.service/package.homechoice/test_ObjServiceInovo.py
Result:
27 passed in 0.14sDeployment: UAT and Status: Safe for testing.What this validates:
data_api_batch_history does not exist.Connect), including lock-denied short-circuit handling.Current limitation of this test run: