Refactor error logging in Salesforce sinks to include sink name and r…

…esponse details

- Updated log_error_message method to accept sink_name and log more detailed error messages, including response status code, URL, and request body.
- Adjusted calls to log_error_message across ContactsSink, CampaignSink, and FallbackSink to pass the sink name for better context in error logs.

target_salesforce_v3/client.py

@@ -115,7 +115,7 @@ def check_salesforce_limits(self, response):
     )
     def _request(
         self, http_method, endpoint, params=None, request_data=None, headers=None
-    ) -> requests.PreparedRequest:
+    ) -> requests.Response:
         """Prepare a request object."""
         url = self.url(endpoint)
         headers = self.http_headers
@@ -141,7 +141,7 @@ def _request(
         self.validate_response(response)
         return response
 
-    def request_api(self, http_method, endpoint=None, params=None, request_data=None, headers=None):
+    def request_api(self, http_method, endpoint=None, params=None, request_data=None, headers=None) -> requests.Response:
         """Request records from REST endpoint(s), returning response records."""
         resp = self._request(http_method, endpoint, params, request_data, headers)
         self.check_salesforce_limits(resp)
@@ -443,11 +443,11 @@ def map_only_empty_fields(self, mapping, sobject_name, lookup_field):
         return mapping
 
 
-    def log_error_message(self, response: PreparedRequest, e: Exception):
+    def log_error_message(self, sink_name: str, response: requests.Response, e: Exception):
         try:
             response_content = response.json()
         except ValueError:
             response_content = response.text if hasattr(response, 'text') else str(response)
 
-        error_message = f"Error: {e}, Response: {response_content}"
-        self.logger.exception(error_message)
+        error_message = f"[{response.status_code}] Sink: {sink_name} | Error: {e}, URL: {response.url}, Body: {response.request.body}, Response: {response_content}"
+        self.logger.exception(error_message)

target_salesforce_v3/sinks.py

@@ -254,7 +254,7 @@ def upsert_record(self, record, context):
                     self.assign_to_topic(id,self.topics)
                 return id, True, state_updates
             except Exception as e:
-                self.log_error_message(response, e)
+                self.log_error_message(self.name, response, e)
                 raise e
 
     def validate_response(self, response):
@@ -320,7 +320,7 @@ def assign_to_topic(self,contact_id,topics:list) -> None:
                 # Means it's already in the topic
                 if "DUPLICATE_VALUE" in str(e):
                     return
-                self.log_error_message(response, e)
+                self.log_error_message(self.name, response, e)
                 raise e
 
     def assign_to_campaign(self,contact_id,campaigns:list) -> None:
@@ -375,7 +375,7 @@ def assign_to_campaign(self,contact_id,campaigns:list) -> None:
                 id = data.get("id")
                 self.logger.info(f"CampaignMember created with id: {id}")
             except Exception as e:
-                self.log_error_message(response, e)
+                self.log_error_message(self.name, response, e)
                 raise e
 
 
@@ -722,7 +722,7 @@ def upsert_record(self, record, context):
             self.logger.info(f"{self.name} created with id: {id}")
             return id, True, state_updates
         except Exception as e:
-            self.log_error_message(response, e)
+            self.log_error_message(self.name, response, e)
             raise e
 
 
@@ -1011,7 +1011,7 @@ def upsert_record(self, record, context):
                 self.logger.info(f"{object_type} updated with id: {id}")
                 return id, True, state_updates
             except Exception as e:
-                self.log_error_message(response, e)
+                self.log_error_message(self.name, response, e)
 
         if len(possible_update_fields) > 0:
             for id_field in possible_update_fields:
@@ -1036,7 +1036,7 @@ def upsert_record(self, record, context):
             return id, True, state_updates
         except Exception as e:
             if "INVALID_FIELD_FOR_INSERT_UPDATE" not in str(e):
-                self.log_error_message(response, e)
+                self.log_error_message(self.name, response, e)
                 raise e
             try:
                 fields = json.loads(str(e))[0]['fields']
@@ -1058,7 +1058,7 @@ def upsert_record(self, record, context):
                 self.logger.info(f"{object_type} created with id: {id}")
                 return id, True, state_updates
             except Exception as e:
-                self.log_error_message(response, e)
+                self.log_error_message(self.name, response, e)
                 raise e