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 48a4f3ae2e
commit 4e40d5e0ad
3 changed files with 52 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
connector-superoffice/superoffice_client.py
View File

@@ -80,6 +80,8 @@ class SuperOfficeClient:
resp = requests.post(url, headers=self.headers, json=payload)
elif method == "PUT":
resp = requests.put(url, headers=self.headers, json=payload)
elif method == "PATCH":
resp = requests.patch(url, headers=self.headers, json=payload)
# 401 Handling
if resp.status_code == 401 and retry:
@@ -109,6 +111,9 @@ class SuperOfficeClient:
def _put(self, endpoint, payload):
return self._request_with_retry("PUT", endpoint, payload)
def _patch(self, endpoint, payload):
return self._request_with_retry("PATCH", endpoint, payload)
def _post(self, endpoint, payload):
return self._request_with_retry("POST", endpoint, payload)
@@ -196,53 +201,45 @@ class SuperOfficeClient:
def update_entity_udfs(self, entity_id: int, entity_type: str, udf_data: dict):
"""
Updates UDFs for a given entity (Contact or Person).
Updates UDFs for a given entity (Contact or Person) using PATCH.
entity_type: 'Contact' or 'Person'
udf_data: {ProgId: Value}
"""
endpoint = f"{entity_type}/{entity_id}"
# 1. GET existing
existing_entity = self._get(endpoint)
if not existing_entity:
logger.error(f"❌ Failed to retrieve existing {entity_type} {entity_id} for UDF update.")
return False
if "UserDefinedFields" not in existing_entity:
existing_entity["UserDefinedFields"] = {}
# Construct PATCH payload
payload = {
"UserDefinedFields": udf_data
}
# 2. Merge payload
existing_entity["UserDefinedFields"].update(udf_data)
logger.info(f"Patching {entity_type} {entity_id} UDFs: {udf_data}...")
logger.info(f"Updating {entity_type} {entity_id} UDFs: {udf_data}...")
# 3. PUT update
result = self._put(endpoint, existing_entity)
# PATCH update
result = self._patch(endpoint, payload)
return bool(result)
def update_person_position(self, person_id: int, position_id: int):
"""
Updates the standard 'Position' field of a Person.
Updates the standard 'Position' field of a Person using PATCH.
"""
endpoint = f"Person/{person_id}"
# 1. GET existing
existing_person = self._get(endpoint)
if not existing_person:
logger.error(f"❌ Failed to retrieve Person {person_id} for Position update.")
return False
# 2. Check current value to avoid redundant updates
current_pos = existing_person.get("Position", {})
if current_pos and str(current_pos.get("Id")) == str(position_id):
logger.info(f"Person {person_id} Position already set to {position_id}. Skipping.")
return True
# Construct PATCH payload
payload = {
"Position": {"Id": int(position_id)}
}
logger.info(f"Patching Person {person_id} Position to ID {position_id}...")
# PATCH update
result = self._patch(endpoint, payload)
return bool(result)
# 3. Update
existing_person["Position"] = {"Id": int(position_id)}
logger.info(f"Updating Person {person_id} Position to ID {position_id}...")
# 4. PUT
result = self._put(endpoint, existing_person)
def patch_contact(self, contact_id: int, patch_data: dict):
"""
Generic PATCH for Contact entity.
"""
endpoint = f"Contact/{contact_id}"
logger.info(f"Patching Contact {contact_id} with data keys: {list(patch_data.keys())}...")
result = self._patch(endpoint, patch_data)
return bool(result)