Moving zoho auth into main file
This commit is contained in:
parent
c7f3d39a27
commit
fe506b0830
56
doors.py
56
doors.py
|
@ -91,11 +91,63 @@ def blink_leds():
|
||||||
lock_door()
|
lock_door()
|
||||||
|
|
||||||
|
|
||||||
|
import datetime
|
||||||
|
import json
|
||||||
|
from http import HTTPStatus
|
||||||
|
from urllib import request
|
||||||
|
from urllib.parse import urlencode
|
||||||
|
from urllib.error import HTTPError
|
||||||
|
|
||||||
|
class ZohoToken:
|
||||||
|
"""
|
||||||
|
Provides fresh API access tokens from Zoho
|
||||||
|
"""
|
||||||
|
def __init__(self, client_id, client_secret, refresh_token):
|
||||||
|
self.client_id = client_id
|
||||||
|
self.client_secret = client_secret
|
||||||
|
self.refresh_token = refresh_token
|
||||||
|
self._access_token = None
|
||||||
|
self.expires_at = None
|
||||||
|
|
||||||
|
def update_token(self):
|
||||||
|
url = "https://accounts.zoho.com/oauth/v2/token?%s" % urlencode({
|
||||||
|
"refresh_token": self.refresh_token,
|
||||||
|
"client_id": self.client_id,
|
||||||
|
"client_secret": self.client_secret,
|
||||||
|
"grant_type": "refresh_token",
|
||||||
|
})
|
||||||
|
req = request.Request(url, method="POST")
|
||||||
|
with request.urlopen(req) as f:
|
||||||
|
data = json.loads(f.read().decode())
|
||||||
|
self._access_token = data["access_token"]
|
||||||
|
self.expires_at = datetime.datetime.now() + datetime.timedelta(seconds=int(data["expires_in"]))
|
||||||
|
|
||||||
|
def get_token(self):
|
||||||
|
if self._access_token is None or self.expires_at <= datetime.datetime.now():
|
||||||
|
self.update_token()
|
||||||
|
return self._access_token
|
||||||
|
|
||||||
|
def authenticate(token, key):
|
||||||
|
"""
|
||||||
|
Authenticate a key against the ZOHO CRM API
|
||||||
|
"""
|
||||||
|
url = "https://www.zohoapis.com/crm/v2/Contacts/search?%s" % urlencode({
|
||||||
|
"criteria": "(Key_ID:equals:%s)" % key,
|
||||||
|
})
|
||||||
|
req = request.Request(url, method="GET", headers={
|
||||||
|
"Authorization": "Zoho-oauthtoken %s" % token.get_token(),
|
||||||
|
})
|
||||||
|
print("Pinging server with key: %s (%s)" % (key, url))
|
||||||
|
try:
|
||||||
|
with request.urlopen(req) as response:
|
||||||
|
return response.status == HTTPStatus.OK
|
||||||
|
except HTTPError as e:
|
||||||
|
print("Error pinging server: %s" % e)
|
||||||
|
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
import time
|
import time
|
||||||
import datetime
|
|
||||||
from zoho_auth import ZohoToken, authenticate
|
|
||||||
from slacker import Slacker
|
from slacker import Slacker
|
||||||
#############################################
|
#############################################
|
||||||
# #
|
# #
|
||||||
|
|
|
@ -3,7 +3,7 @@ import unittest
|
||||||
from freezegun import freeze_time
|
from freezegun import freeze_time
|
||||||
from unittest.mock import patch, MagicMock
|
from unittest.mock import patch, MagicMock
|
||||||
from urllib.error import HTTPError
|
from urllib.error import HTTPError
|
||||||
from zoho_auth import ZohoToken, authenticate
|
from doors import ZohoToken, authenticate
|
||||||
|
|
||||||
class TestZohoToken(unittest.TestCase):
|
class TestZohoToken(unittest.TestCase):
|
||||||
"""Test the ZohoToken class"""
|
"""Test the ZohoToken class"""
|
||||||
|
|
48
zoho_auth.py
48
zoho_auth.py
|
@ -1,48 +0,0 @@
|
||||||
import datetime
|
|
||||||
import json
|
|
||||||
from http import HTTPStatus
|
|
||||||
from urllib import request
|
|
||||||
from urllib.parse import urlencode
|
|
||||||
from urllib.error import HTTPError
|
|
||||||
|
|
||||||
class ZohoToken:
|
|
||||||
def __init__(self, client_id, client_secret, refresh_token):
|
|
||||||
self.client_id = client_id
|
|
||||||
self.client_secret = client_secret
|
|
||||||
self.refresh_token = refresh_token
|
|
||||||
self._access_token = None
|
|
||||||
self.expires_at = None
|
|
||||||
|
|
||||||
def update_token(self):
|
|
||||||
url = "https://accounts.zoho.com/oauth/v2/token?%s" % urlencode({
|
|
||||||
"refresh_token": self.refresh_token,
|
|
||||||
"client_id": self.client_id,
|
|
||||||
"client_secret": self.client_secret,
|
|
||||||
"grant_type": "refresh_token",
|
|
||||||
})
|
|
||||||
req = request.Request(url, method="POST")
|
|
||||||
with request.urlopen(req) as f:
|
|
||||||
data = json.loads(f.read().decode())
|
|
||||||
self._access_token = data["access_token"]
|
|
||||||
self.expires_at = datetime.datetime.now() + datetime.timedelta(seconds=int(data["expires_in"]))
|
|
||||||
|
|
||||||
def get_token(self):
|
|
||||||
if self._access_token is None or self.expires_at <= datetime.datetime.now():
|
|
||||||
self.update_token()
|
|
||||||
return self._access_token
|
|
||||||
|
|
||||||
def authenticate(token, key):
|
|
||||||
url = "https://www.zohoapis.com/crm/v2/Contacts/search?%s" % urlencode({
|
|
||||||
"criteria": "(Key_ID:equals:%s)" % key,
|
|
||||||
})
|
|
||||||
req = request.Request(url, method="GET", headers={
|
|
||||||
"Authorization": "Zoho-oauthtoken %s" % token.get_token(),
|
|
||||||
})
|
|
||||||
print("Pinging server with key: %s (%s)" % (key, url))
|
|
||||||
try:
|
|
||||||
with request.urlopen(req) as response:
|
|
||||||
return response.status == HTTPStatus.OK
|
|
||||||
except HTTPError as e:
|
|
||||||
print("Error pinging server: %s" % e)
|
|
||||||
|
|
||||||
return False
|
|
Loading…
Reference in New Issue