fureunoir/schon
fureunoir/schon
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Features: 1) Add support for authorization code grant type in the Amo gateway.

Browse source 
Fixes: 1) Improve validation logic to ensure only one default CRM provider within the model.

Extra: 1) Replace CRM object string representation to use `name` instead of `crm_lead_id`; 2) Refactor token caching to improve reliability and readability.
This commit is contained in:
Egor Pavlovich Gorbunov 2025-09-07 03:03:29 +03:00
parent 3769482f56
commit 0ca0756e50
2 changed files with 19 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
core/crm/amo/gateway.py
View file

@ -25,11 +25,9 @@ class AmoCRM:
self.client_id = self.instance.authentication.get("client_id")
self.client_secret = self.instance.authentication.get("client_secret")
self.authorization_code = self.instance.authentication.get("authorization_code")
self.refresh_token = cache.get("amo_refresh_token")
self.redirect_uri = self.instance.attributes.get("redirect_uri")
self.pipeline_id = self.instance.attributes.get("pipeline_id")
self.stage_map = self.instance.attributes.get("stage_map")
self.responsible_user_id = self.instance.attributes.get("responsible_user_id")
self.fns_api_key = self.instance.attributes.get("fns_api_key")
@ -39,7 +37,7 @@ class AmoCRM:
self.base,
self.client_id,
self.client_secret,
self.redirect_uri,
self.authorization_code,
self.fns_api_key,
]
):
@ -53,17 +51,20 @@ class AmoCRM:
payload = {
"client_id": self.client_id,
"client_secret": self.client_secret,
"grant_type": "refresh_token",
"refresh_token": self.refresh_token,
"redirect_uri": self.redirect_uri,
}
if self.refresh_token:
payload["grant_type"] = "refresh_token"
payload["refresh_token"] = self.refresh_token
else:
payload["grant_type"] = "authorization_code"
payload["code"] = self.authorization_code
r = requests.post(f"{self.base}/oauth2/access_token", json=payload, timeout=15)
r.raise_for_status()
data = r.json()
self._cached_token = data["access_token"]
self._cached_expiry = time.time() + int(data.get("expires_in", 900))
self.refresh_token = data.get("refresh_token", self.refresh_token)
return self._cached_token
self.access_token = data["access_token"]
cache.set("amo_refresh_token", data["refresh_token"], 604800)
self.refresh_token = data["refresh_token"]
return self.access_token
def _headers(self) -> dict:
return {"Authorization": f"Bearer {self._token()}", "Content-Type": "application/json"}

8
core/models.py
View file

@ -1969,10 +1969,14 @@ class CustomerRelationshipManagementProvider(ExportModelOperationsMixin("crm_pro
default = BooleanField(default=False)
def __str__(self) -> str:
return self.crm_lead_id
return self.name
def save(self, **kwargs):
if self.objects.filter(default=True).exists():
if self.default:
qs = type(self).objects.all()
if self.pk:
qs = qs.exclude(pk=self.pk)
if qs.filter(default=True).exists():
raise ValueError(_("you can only have one default CRM provider"))
super().save(**kwargs)