Architectural Deep-Dive: Origami Risk Bulk API
Integrating enterprise Risk Management Information Systems (RMIS) like Origami Risk presents unique engineering challenges. While the Origami platform is incredibly flexible, its backend API exhibits specific architectural behaviors that developers must design around when scaling automated data pipelines (such as Location, Vehicle, or Policy syncs).
This post explores the engineering lessons learned from scaling an Origami integration, navigating its transactional database behavior, and building a resilient, self-healing Python orchestration layer.
1. The Anatomy of an Origami "Upsert" Failure
A common point of confusion when working with the /api/Domain/BulkUpsert endpoint is how Origami evaluates an "Upsert."
The endpoint allows you to pass a URL query string string parameter called matchFields (e.g., ?matchFields=VehicleNumber,Make,Model,VIN). If your input dataset passes fields in matchFields that are attributes rather than immutable unique keys, the API engine executes a strict AND lookup.
If even a single attribute has changed in your source system (e.g., a vehicle changed locations), the engine finds zero exact matches across all specified fields. It assumes this is a brand-new record and attempts a SQL INSERT. This instantly triggers a hard database constraint violation:
Cannot insert duplicate key row in object 'dbo.Vehicles' with unique index 'UIDX_Vehicles_VehicleNumber'.
The Engineering Fix
The matchFields parameter must only contain the minimum natural or business key that uniquely identifies the entity in the platform configuration (e.g., matchFields=VehicleNumber or matchFields=Location,Value,PeriodStart). This forces Origami to locate the correct target record and cleanly execute an underlying SQL UPDATE statement on the changing attributes.
2. Gateway Timeout vs. Database Lock Escalation
When scaling up batch transaction sizes from small blocks of 25 records to thousands of records per API call, pipelines often encounter abrupt network drops:
urllib3.exceptions.ProtocolError: ('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))
This error is rarely a client-side timeout. Instead, it is a gateway drop caused by a conflict between two system layers:
-
The Web Gateway: Origami’s web application servers or load balancers enforce rigid request timeout limits (often around 30 to 60 seconds).
-
The Relational Database Layer: When processing large payloads, Origami must perform heavy business logic—such as evaluating unique index constraints, parsing custom fields, and triggering system events (
fireEvents=true)—all within a single transactional block.
If a 1,000-record batch takes 90 seconds for the database engine to parse and commit, the web gateway will cut the connection socket mid-flight, dropping the connection before the client can receive a response.
The "Queue" Effect and Lock Contention
A compounding issue is transactional queuing. Even if a 200-row batch successfully commits, the immediate next batch might time out.
This happens because the database escalates its row-level locks to a table-level lock to clear its indexing buffers from the first transaction. The second request gets trapped in a lock wait queue. By the time the table lock releases, the web gateway's idle timer has expired.
3. Designing a Resilient Python Pipeline for Origami
To build an enterprise-grade pipeline that can run unattended via agents like Informatica, we implemented a defensive Python wrapper utilizing three core patterns:
A. The "Sweet Spot" Strategy (Micro-Batching)
Instead of overwhelming the backend transaction threads with batches of 1,000 or dealing with the slow performance of 25, testing mapped a stable throughput envelope at 100 records per batch, paired with a 45-second read timeout, and a 2.5-second sleep delay. This minor cooldown window gives Origami’s indexing engine time to clear transaction locks before the next thread hits.
B. Graceful Handling of Mixed-Status HTTP 400 Responses
If one record in a batch fails a validation check (e.g., a duplicate VIN error), Origami’s API handler returns a global HTTP 400 Bad Request status, rather than an HTTP 200.
Using standard code shortcuts like response.raise_for_status() here will crash your entire run loop, hiding the fact that other records in that same batch may have processed successfully. The pipeline must intercept HTTP 400 responses, parse the underlying JSON response array, identify individual IsSuccessful: false records, and keep moving.
C. Implementing an Automated Circuit Breaker & Dead-Letter Queue (DLQ)
When interacting with a volatile cloud backend, your code must protect the state of the data. The refactored pipeline integrates two specific guardrails:
-
Automated Dead-Letter Queue (DLQ): If a batch exhausts its inner retry attempts due to network drops, the raw 100-record block is dumped out to an isolation file (e.g.,
failed_batch_5.json). This ensures zero data loss and allows for isolated reprocessing later. -
The Circuit Breaker: The system tracks global stability. If it encounters 5 consecutive timeouts or 10 cumulative timeouts across the entire run, it raises a terminal
RuntimeErrorto stop the script from infinite looping against a hanging server.
Final Takeaways
When integrating with enterprise platforms like Origami Risk:
-
Micro-transactions win: A steady sequence of smaller, highly optimized transactions is faster and more stable than huge, bulky batches that risk triggering lock escalation.
-
Isolate your match logic: Keep your URL
matchFieldslimited to immutable business keys. -
Code defensively: Treat every API response as potentially volatile. Build dead-letter files and execution limits directly into your loop structures.
Sample Code:
def origmai_upload_data(upload_type):
normalized_type = str(upload_type).strip().lower()
upload_configs = {
"vehicle": {
"query": VEHICLE_QUERY,
"url": f"{ORIGAMI_BASE_URL}/api/Vehicle/BulkUpsert?matchFields=VehicleNumber&fireEvents=true"
},
"location_ee": {
"query": LOCATION_EE_QUERY,
"url": f"{ORIGAMI_BASE_URL}/api/LocationValue/BulkUpsert?matchFields=Location,Value,PeriodStart&fireEvents=true"
},
"employee": {
"query": EMPLOYEE_QUERY,
"url": f"{ORIGAMI_BASE_URL}/api/Employee/BulkUpsert?matchFields=EmployeeNumber&fireEvents=true"
}
}
if normalized_type not in upload_configs:
print(f"❌ Invalid argument '{upload_type}' passed to origmai_upload_data")
sys.exit(-1)
config = upload_configs[normalized_type]
df_results = fun_select_query(config["query"], "Y")
if df_results.empty:
print(f"No records found for {normalized_type}. Exiting execution.")
return
print(f"🚀 Records found for {normalized_type} payload. Normalizing schema...")
all_records = []
if normalized_type == "location_ee":
payload_df = df_results.copy()
if payload_df["date"].dtype == "datetime64[ns]":
payload_df["PeriodStart"] = payload_df["date"].dt.strftime("%Y-%m-%d")
else:
payload_df["PeriodStart"] = payload_df["date"].astype(str)
all_records = [
{
"Location": row["location"],
"Value": row["ValueName"],
"PeriodStart": row["PeriodStart"],
"ActualAmount": float(row["ActualAmount"]),
"Status": "A",
}
for _, row in payload_df.iterrows()
]
elif normalized_type == "vehicle":
all_records = [
{
"VehicleNumber": row["vehiclenumber"],
"Make": row["make"],
"Model": row["model"],
"Year": row["year"],
"VIN": row["vin"],
"Location": row["location"],
"RegisteredState": map_state_to_code(row["registered_state"]),
"Status": row["status"],
"CustomText1": row["customtext1"]
}
for _, row in df_results.iterrows()
]
elif normalized_type == "employee":
all_records = [
{
"Country": row["Country"],
"EmployeeNumber": row["EmployeeNumber"],
"FirstName": row["FirstName"],
"LastName": row["LastName"],
"MiddleName": row["MiddleName"],
"Email1": row["Email1"],
"Email2": row["Email2"],
"WorkPhone": row["WorkPhone"],
"EmploymentStatus": row["EmploymentStatus"],
"HireDate": row["HireDate"],
"TerminationDate": row["TerminationDate"],
"HICN": row["HICN"],
"Location": row["Location"],
"Occupation": row["Occupation"],
"Department": row["Department"],
"JobClassificationID": "",
"WageRateType": row["WageRateType"],
"SupervisorEmployee": row["SupervisorEmployee"],
"CustomText6": row["CustomText6"],
"CustomNumber1": row["CustomNumber1"],
"CustomText4": row["CustomText4"],
"State": row["State"]
}
for _, row in df_results.iterrows()
]
with open(f"{INFA_SRC_FOLDER}{normalized_type}_upload_test.json", "w", encoding="utf-8") as f:
f.write(json.dumps(all_records, indent=2))
CHUNK_SIZE = 100
payload_batches = [
all_records[i : i + CHUNK_SIZE]
for i in range(0, len(all_records), CHUNK_SIZE)
]
print(f"Total records: {len(all_records)}. Split into {len(payload_batches)} batches.")
bearer_token = origmai_auth_token()
headers = {
"Authorization": f"Bearer {bearer_token}",
"Content-Type": "application/json",
"Accept": "application/json",
}
all_responses = []
consecutive_timeouts = 0
total_timeouts = 0
for index, batch in enumerate(payload_batches):
batch_num = index + 1
print(f"Uploading batch {batch_num}/{len(payload_batches)}...")
batch_was_processed = False
for attempt in range(2):
response = None
try:
response = requests.post(config["url"], headers=headers, json=batch, timeout=(10, 45))
if response.status_code == 401:
if attempt == 0:
print(f"🔑 Token expired on batch {batch_num}. Regenerating fresh token...")
bearer_token = origmai_auth_token()
headers["Authorization"] = f"Bearer {bearer_token}"
continue
else:
print(f"❌ Token retry failed with a consecutive 401. Moving to next batch.")
response_payload = None
if response.status_code == 400:
try:
response_payload = response.json()
print(f"⚠️ Batch {batch_num} returned HTTP 400 (Validation Errors Encountered).")
except ValueError:
pass
if response_payload is None:
response.raise_for_status()
response_payload = response.json()
if isinstance(response_payload, list):
failed_records = [
rec for rec in response_payload if not rec.get("IsSuccessful", True)
]
if failed_records:
print(f"⚠️ Batch {batch_num} partially failed! {len(failed_records)} record(s) encountered validation errors.")
for rec in failed_records:
print(f" - Record {rec.get('RecordNumber')}: {rec.get('Message')}")
else:
print(f"✅ Batch {batch_num} uploaded successfully (All records IsSuccessful=True).")
else:
if response_payload.get("status") == "fail":
print(f"❌ Internal Origami failure on batch {batch_num}: {response_payload.get('exception')}")
all_responses.append({
"batch_number": batch_num,
"status": "processed",
"data": response_payload,
})
consecutive_timeouts = 0
batch_was_processed = True
break
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as timeout_err:
if attempt == 0:
print(f"🔄 Intermittent network glitch on batch {batch_num} (Attempt 1). Retrying...")
time.sleep(1.0)
continue
total_timeouts += 1
consecutive_timeouts += 1
print(f"⏰ Batch {batch_num} TIMED OUT. Total Timeouts: {total_timeouts}/10 | Consecutive: {consecutive_timeouts}/5")
dlq_filename = f"{INFA_SRC_FOLDER}{normalized_type}_failed_batch_{batch_num}.json"
with open(dlq_filename, "w", encoding="utf-8") as f:
json.dump(batch, f, indent=2)
print(f"💾 Saved raw chunk data to recovery file: {dlq_filename}")
all_responses.append({
"batch_number": batch_num,
"status": "failed_timeout",
"error": str(timeout_err),
})
if consecutive_timeouts >= 5:
raise RuntimeError(f"⛔ Aborted: Hit threshold of {consecutive_timeouts} consecutive network timeouts.")
if total_timeouts >= 10:
raise RuntimeError(f"⛔ Aborted: Hit threshold of {total_timeouts} global runtime network timeouts.")
batch_was_processed = True
break
except requests.exceptions.RequestException as e:
if attempt == 1 or (response is not None and response.status_code != 401):
print(f"Critical execution error uploading batch {batch_num}: {e}")
all_responses.append({
"batch_number": batch_num,
"status": "failed",
"error": str(e),
})
batch_was_processed = True
break
if batch_was_processed and consecutive_timeouts == 0:
time.sleep(2.5)
else:
time.sleep(0.5)
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
log_filename = f"{INFA_SRC_FOLDER}{normalized_type}_upload_response_{timestamp}.json"
with open(log_filename, "w", encoding="utf-8") as f:
f.write(json.dumps(all_responses, indent=2))
print(f"📋 Global run audit log saved to: {log_filename}")