Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor API calls to use PATCH instead of PUT for cleaner updates (SuperOffice Certification)

Browse Source
This commit is contained in:
Floke
2026-02-25 17:21:36 +00:00
parent 69bc917916
commit f94f7ea4dd
3 changed files with 52 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
connector-superoffice/worker.py
View File

@@ -214,8 +214,7 @@ def process_job(job, so_client: SuperOfficeClient):
logger.error(f"Failed to fetch contact {contact_id}: {e}")
return "RETRY"
dirty_standard = False
dirty_udfs = False
contact_patch = {}
if "UserDefinedFields" not in contact_data: contact_data["UserDefinedFields"] = {}
@@ -233,8 +232,7 @@ def process_job(job, so_client: SuperOfficeClient):
if str(current_val) != str(vertical_id):
logger.info(f"Change detected: Vertical {current_val} -> {vertical_id}")
contact_data["UserDefinedFields"][udf_key] = str(vertical_id)
dirty_udfs = True
contact_patch.setdefault("UserDefinedFields", {})[udf_key] = str(vertical_id)
except Exception as e:
logger.error(f"Vertical sync error: {e}")
@@ -252,26 +250,23 @@ def process_job(job, so_client: SuperOfficeClient):
if "Postal" not in addr_obj or addr_obj["Postal"] is None: addr_obj["Postal"] = {}
if "Street" not in addr_obj or addr_obj["Street"] is None: addr_obj["Street"] = {}
def update_addr_field(field_name, new_val, log_name):
nonlocal dirty_standard
def update_addr_patch(field_name, new_val, log_name):
if new_val:
for type_key in ["Postal", "Street"]:
cur = addr_obj[type_key].get(field_name, "")
if cur != new_val:
logger.info(f"Change detected: {type_key} {log_name} '{cur}' -> '{new_val}'")
addr_obj[type_key][field_name] = new_val
dirty_standard = True
contact_patch.setdefault("Address", {}).setdefault(type_key, {})[field_name] = new_val
update_addr_field("City", ce_city, "City")
update_addr_field("Address1", ce_street, "Street")
update_addr_field("Zipcode", ce_zip, "Zip")
update_addr_patch("City", ce_city, "City")
update_addr_patch("Address1", ce_street, "Street")
update_addr_patch("Zipcode", ce_zip, "Zip")
if ce_vat:
current_vat = contact_data.get("OrgNr", "")
if current_vat != ce_vat:
logger.info(f"Change detected: VAT '{current_vat}' -> '{ce_vat}'")
contact_data["OrgNr"] = ce_vat
dirty_standard = True
contact_patch["OrgNr"] = ce_vat
# --- C. AI Openers Sync ---
ce_opener = provisioning_data.get("opener")
@@ -281,25 +276,25 @@ def process_job(job, so_client: SuperOfficeClient):
current_opener = contact_data["UserDefinedFields"].get(settings.UDF_OPENER, "")
if current_opener != ce_opener:
logger.info("Change detected: Primary Opener")
contact_data["UserDefinedFields"][settings.UDF_OPENER] = ce_opener
dirty_udfs = True
contact_patch.setdefault("UserDefinedFields", {})[settings.UDF_OPENER] = ce_opener
if ce_opener_secondary and ce_opener_secondary != "null":
current_opener_sec = contact_data["UserDefinedFields"].get(settings.UDF_OPENER_SECONDARY, "")
if current_opener_sec != ce_opener_secondary:
logger.info("Change detected: Secondary Opener")
contact_data["UserDefinedFields"][settings.UDF_OPENER_SECONDARY] = ce_opener_secondary
dirty_udfs = True
contact_patch.setdefault("UserDefinedFields", {})[settings.UDF_OPENER_SECONDARY] = ce_opener_secondary
# --- D. Apply Updates (Single Transaction) ---
if dirty_standard or dirty_udfs:
logger.info(f"Pushing combined updates for Contact {contact_id}...")
# --- D. Apply Updates (Single PATCH Transaction) ---
if contact_patch:
logger.info(f"Pushing combined PATCH updates for Contact {contact_id}...")
try:
so_client._put(f"Contact/{contact_id}", contact_data)
logger.info("✅ Contact Update Successful.")
so_client.patch_contact(contact_id, contact_patch)
logger.info("✅ Contact Update Successful (PATCH).")
except Exception as e:
logger.error(f"Failed to update Contact {contact_id}: {e}")
raise
else:
logger.info(f"No changes detected for Contact {contact_id}. Skipping update.")
# 2d. Sync Person Position (Role) - if Person exists
role_name = provisioning_data.get("role_name")