refactor: make sipgate client frappe free (GDE-80)

sipgate/install.py

@@ -18,10 +18,12 @@ def make_custom_fields():
 		{
 			"fieldname": "sipgate_id",
 			"fieldtype": "Data",
-			"label": "Sipgate Id",
+			"label": "Sipgate ID",
 			"insert_after": "unsubscribed",
 			"read_only": 1,
 			"translatable": 0,
+			"no_copy": 1,
+			"unique": 1,
 		}
 	],
 }

sipgate/patches.txt

@@ -0,0 +1,2 @@
+execute:frappe.db.sql("UPDATE tabContact SET sipgate_id = NULL") # remove old duplicate IDs to enable unique constraint
+execute:from sipgate.install import after_install; after_install()

sipgate/sipgate/doctype/sipgate_settings/sipgate_settings.py

@@ -1,118 +1,75 @@
 # Copyright (c) 2022, ALYF GmbH and contributors
 # For license information, please see license.txt
 
+from typing import Optional, Union
+
 import frappe
-import requests
 from frappe.model.document import Document
-from requests.auth import HTTPBasicAuth
-from frappe import _
-from typing import Union
+from frappe.contacts.doctype.contact.contact import Contact
+
+from sipgate.sipgate_client import SipgateClient
 
 
 class SipgateSettings(Document):
 	pass
 
 
-def sync_to_sipgate(doc=None, method=None):
+def sync_to_sipgate(doc: Contact, method: Optional[str] = None):
 	sipgate_settings = frappe.get_single("Sipgate Settings")
 	if not sipgate_settings.enabled:
 		return
 
-	sipgate = Sipgate(sipgate_settings, doc)
-
-	if not doc.sipgate_id:
-		sipgate.upload()
-		return
-
-	sipgate.update()
-
-
-class Sipgate:
-	def __init__(self, sipgate_settings: SipgateSettings, contact: object) -> None:
-		self.sipgate_settings = sipgate_settings
-		self.auth = HTTPBasicAuth(
-			self.sipgate_settings.token_id, self.sipgate_settings.get_password("token")
-		)
-		self.contact = contact
-
-	def upload(self) -> None:
-		payload = self.get_payload()
-		try:
-			response = requests.post(
-				url=f"{self.sipgate_settings.url}/contacts",
-				json=payload,
-				auth=self.auth,
-			)
-			response.raise_for_status()
-			if response.status_code == 201:
-				id = self.get_sipgate_id()
-				frappe.db.set_value("Contact", self.contact.name, "sipgate_id", id)
-		except Exception:
-			frappe.msgprint(_("Couldn't sync contact to Sipgate."))
-
-	def update(self) -> None:
-		payload = self.get_payload()
-		try:
-			response = requests.put(
-				url=f"{self.sipgate_settings.url}/contacts/{self.contact.sipgate_id}",
-				json=payload,
-				auth=self.auth,
-			)
-			response.raise_for_status()
-		except Exception:
-			frappe.msgprint(_("Couldn't sync contact to Sipgate."))
-
-	def get_sipgate_id(self) -> Union[str, None]:
-		response = requests.get(
-			url=f"{self.sipgate_settings.url}/contacts",
-			data={"phonenumbers": self.get_contact_number()},
-			auth=self.auth,
-		)
-
-		if response.status_code == 200:
-			response = response.json().get("items", [])
-			if response:
-				return response[0].get("id")
-
-		return None
-
-	def get_contact_number(self) -> Union[str, None]:
-		if self.contact.phone_nos:
-			return self.contact.phone_nos[0].phone
-
-		return None
-
-	def get_payload(self) -> dict:
-		contact_dict = {
-			"name": f"{self.contact.get('first_name', '')} {self.contact.get('last_name', '')}",
-			"family": self.contact.get("last_name", ""),
-			"given": self.contact.get("first_name", ""),
-			"emails": [
-				{
-					"email": email.email_id,
-					"type": [is_primary_email(email)],
-				}
-				for email in self.contact.email_ids
-				if email
-			],
-			"numbers": [
-				{
-					"number": phone.phone,
-					"type": [is_primary_phone(phone)],
-				}
-				for phone in self.contact.phone_nos
-				if phone
-			],
-			"scope": "SHARED",
-		}
-
-		if self.contact.company_name:
-			contact_dict.update({"organization": [[self.contact.company_name]]})
-
-		if self.contact.get("sipgate_id"):
-			contact_dict.update({"id": self.contact.get("sipgate_id")})
-
-		return contact_dict
+	payload = get_payload(doc)
+	sipgate = SipgateClient(
+		sipgate_settings.url, sipgate_settings.token_id, sipgate_settings.get_password("token")
+	)
+
+	try:
+		if doc.sipgate_id:
+			sipgate.update(payload)
+		else:
+			sipgate.upload(payload)
+			id = sipgate.get_sipgate_id(get_contact_number(doc), payload.get("name"))
+			frappe.db.set_value(doc.doctype, doc.name, "sipgate_id", id)
+	except Exception:
+		frappe.log_error(frappe.get_traceback())
+
+
+def get_payload(contact: Contact) -> frappe._dict:
+	payload = {
+		"name": f"{contact.get('first_name', '')} {contact.get('last_name', '')}",
+		"family": contact.get("last_name", ""),
+		"given": contact.get("first_name", ""),
+		"emails": [
+			{
+				"email": email.email_id,
+				"type": [is_primary_email(email)],
+			}
+			for email in contact.email_ids
+			if email
+		],
+		"numbers": [
+			{
+				"number": phone.phone,
+				"type": [is_primary_phone(phone)],
+			}
+			for phone in contact.phone_nos
+			if phone
+		],
+		"scope": "SHARED",
+	}
+
+	if contact.company_name:
+		payload.update({"organization": [[contact.company_name]]})
+
+	if contact.get("sipgate_id"):
+		payload.update({"id": contact.get("sipgate_id")})
+
+	return frappe._dict(payload)
+
+
+def get_contact_number(doc) -> Union[str, None]:
+	return doc.phone_nos[0].phone if doc.phone_nos else None
 
 
 def is_primary_phone(phone: object) -> str:
@@ -121,3 +78,4 @@ def is_primary_phone(phone: object) -> str:
 
 def is_primary_email(email: object) -> str:
 	return "primary" if email.is_primary else ""
+

sipgate/sipgate_client.py

@@ -0,0 +1,55 @@
+# Copyright (c) 2022, ALYF GmbH and contributors
+# For license information, please see license.txt
+
+from typing import Union
+
+import requests
+from requests.auth import HTTPBasicAuth
+
+
+class SipgateClient:
+	def __init__(
+		self,
+		sipgate_url: str,
+		sipgate_token_id: str,
+		sipgate_token: str,
+	) -> None:
+		self.sipgate_url = sipgate_url
+		self.sipgate_token_id = sipgate_token_id
+		self.sipgate_token = sipgate_token
+
+	@property
+	def auth(self):
+		return HTTPBasicAuth(self.sipgate_token_id, self.sipgate_token)
+
+	def upload(self, contact: dict) -> None:
+		response = requests.post(
+			url=f"{self.sipgate_url}/contacts",
+			json=contact,
+			auth=self.auth,
+		)
+		response.raise_for_status()
+
+	def update(self, contact: dict) -> None:
+		response = requests.put(
+			url=f"{self.sipgate_url}/contacts/{self.contact.sipgate_id}",
+			json=contact,
+			auth=self.auth,
+		)
+		response.raise_for_status()
+
+	def get_sipgate_id(self, phonenumbers: str, full_name: str) -> Union[str, None]:
+		if not phonenumbers:
+			return None
+
+		response = requests.get(
+			url=f"{self.sipgate_url}/contacts",
+			params={"phonenumbers": [phonenumbers]},
+			auth=self.auth,
+		)
+		response.raise_for_status()
+
+		items = response.json().get("items", [])
+		items = [item for item in items if item.get("name", "") == full_name]
+
+		return items[0].get("id") if items else None