diff --git a/README.md b/README.md index 4377d45..264d3af 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,28 @@ calendly = CalendlyAPI(api_key) - `get_event_details` - Get information about the event - `list_event_invitees` - Get all invitees for a event +### Oauth2 +Getting started with [Calendly Oauth2 API](https://developer.calendly.com/api-docs/YXBpOjU5MTQwNw-o-auth-2-0) . +``` +from calendly import CalendlyOauth2 + +client_id = "" +client_secret = "" +redirect_uri = "" + +oauth2 = CalendlyOauth2( + client_id, + client_secret, + redirect_uri +) +``` + +### Methods +- `authorization_url` - Returns the formatted authorization URL +- `get_access_token` - Send a request to obtain the given access token +- `revoke_access_token` - Send a request to revoke the given access token +- `refresh_access_token` - Send a request to refresh the given access token +- `introspect_access_token` - Send a request to bring details about the given access token ## Issues Feel free to submit issues and enhancement requests. diff --git a/calendly/__init__.py b/calendly/__init__.py index 74f0f61..ca73b4b 100644 --- a/calendly/__init__.py +++ b/calendly/__init__.py @@ -1,3 +1,5 @@ +from .utils.oauth2 import CalendlyOauth2 from .calendly import CalendlyAPI +from .exceptions import CalendlyException, CalendlyOauth2Exception -__all__ = [CalendlyAPI] \ No newline at end of file +__all__ = [CalendlyAPI, CalendlyOauth2, CalendlyException, CalendlyOauth2Exception] \ No newline at end of file diff --git a/calendly/calendly.py b/calendly/calendly.py index b267c68..55948d7 100644 --- a/calendly/calendly.py +++ b/calendly/calendly.py @@ -1,7 +1,10 @@ -from calendly.utils.constants import WEBHOOK, EVENTS, ME, EVENT_TYPE -from calendly.utils.requests import CalendlyReq, CalendlyException +import json from typing import List, MutableMapping -import json + +from calendly.utils.api import CalendlyReq +from calendly.utils.constants import WEBHOOK, EVENTS, ME, EVENT_TYPE +from calendly.exceptions import CalendlyException + class CalendlyAPI(object): @@ -21,7 +24,7 @@ def __init__(self, token: str): """ self.request = CalendlyReq(token) - def create_webhook(self, url: str, scope: str, organization: str, signing_key: str=None, user: str=None, event_types: List[str]=["canceled", "created"]) -> MutableMapping: + def create_webhook(self, url: str, scope: str, organization: str, signing_key: str=None, user: str=None, event_types: List[str]=("canceled", "created")) -> MutableMapping: """ Create a Webhook Subscription @@ -29,6 +32,7 @@ def create_webhook(self, url: str, scope: str, organization: str, signing_key: s url (str): Webhook URL scope (str): Either "organization" or "user" organization (str): Unique reference to the organization that the webhook will be tied to + signing_key (str): A signing key user (str, optional): If scope is set to "user", then user reference is required. event_types (list, optional): List of user events to subscribe to. Defaults to ["canceled", "created"]. @@ -44,8 +48,8 @@ def create_webhook(self, url: str, scope: str, organization: str, signing_key: s 'scope': scope, 'signing_key': signing_key} - if (scope == 'user'): - if (user == None): + if scope == 'user': + if user is None: raise CalendlyException data['user'] = user @@ -131,7 +135,7 @@ def about(self) -> MutableMapping: response = self.request.get(ME) return response.json() - def list_event_types(self, count: int=20, organization: str=None, page_token: str=None, sort: str=None, user_uri: str=None) -> List[MutableMapping]: + def list_event_types(self, count: int=20, organization: str=None, page_token: str=None, sort: str=None, user_uri: str=None) -> MutableMapping: """ Returns all Event Types associated with a specified user. @@ -146,13 +150,13 @@ def list_event_types(self, count: int=20, organization: str=None, page_token: st dict: json decoded response with list of event types """ data = {"count": count} - if (organization): + if organization: data['organization'] = organization - if (page_token): + if page_token: data['page_token'] = page_token - if (sort): + if sort: data['sort'] = sort - if (user_uri): + if user_uri: data['user'] = user_uri response = self.request.get(EVENT_TYPE, data) return response.json() @@ -170,7 +174,7 @@ def get_event_type(self, uuid: str) -> MutableMapping: response = self.request.get(f'{EVENT_TYPE}/' + uuid, data) return response.json() - def list_events(self, count: int=20, organization: str=None, sort: str=None, user_uri: str=None, status: str=None) -> List[MutableMapping]: + def list_events(self, count: int=20, organization: str=None, sort: str=None, user_uri: str=None, status: str=None) -> MutableMapping: """ Returns a List of Events @@ -185,13 +189,13 @@ def list_events(self, count: int=20, organization: str=None, sort: str=None, use dict: json decoded response of list of events. """ data = {'count': count} - if (organization): + if organization: data['organization'] = organization - if (sort): + if sort: data['sort'] = sort - if (user_uri): + if user_uri: data['user'] = user_uri - if (status): + if status: data['status'] = status response = self.request.get(EVENTS, data) return response.json() @@ -239,7 +243,7 @@ def list_event_invitees(self, uuid: str) -> List[MutableMapping]: response = self.request.get(url) return response.json() - def get_all_event_types(self, user_uri: str) -> List[str]: + def get_all_event_types(self, user_uri: str) -> List[MutableMapping]: """ Get all event types by recursively crawling on all result pages. @@ -254,14 +258,14 @@ def get_all_event_types(self, user_uri: str) -> List[str]: data = first['collection'] - while (next_page): + while next_page: page = self.request.get(next_page).json() data += page['collection'] next_page = page['pagination']['next_page'] return data - def get_all_scheduled_events(self, user_uri: str) -> List[str]: + def get_all_scheduled_events(self, user_uri: str) -> List[MutableMapping]: """ Get all scheduled events by recursively crawling on all result pages. @@ -306,5 +310,4 @@ def convert_event_to_original_url(self, event_uri: str, user_uri: str) -> str: if not next_page: break page = self.request.get(next_page).json() - return \ No newline at end of file diff --git a/calendly/exceptions.py b/calendly/exceptions.py new file mode 100644 index 0000000..3a05937 --- /dev/null +++ b/calendly/exceptions.py @@ -0,0 +1,10 @@ +class CalendlyException(Exception): + """Errors corresponding to a misuse of Calendly API""" + + def __init__(self, message=None, details=None): + self.message = message or "" + self.details = details or [] + super(CalendlyException, self).__init__(f"{self.message} - {self.details}") + +class CalendlyOauth2Exception(CalendlyException): + """Errors corresponding to a misuse of CalendlyOauth2 API""" \ No newline at end of file diff --git a/calendly/tests/run_tests.py b/calendly/tests/run_tests.py index 730b0d3..e903776 100644 --- a/calendly/tests/run_tests.py +++ b/calendly/tests/run_tests.py @@ -1,9 +1,13 @@ -from calendly.calendly import CalendlyReq, CalendlyAPI -from calendly.utils import constants -from unittest.mock import MagicMock -import unittest -import json import copy +import json +import unittest +from unittest.mock import MagicMock, patch + +from calendly.calendly import CalendlyAPI +from calendly.exceptions import CalendlyOauth2Exception, CalendlyException +from calendly.utils import constants +from calendly.utils.api import CalendlyReq +from calendly.utils.oauth2 import CalendlyOauth2 # Init test objects mock_token = 'mock_token' @@ -11,6 +15,61 @@ calendly_request = CalendlyReq(mock_token) calendly_client.request = calendly_request + +class TestCalendlyExceptions(unittest.TestCase): + + def test_CalendlyException(self): + message = "SOME MESSAGE" + details = [{}, {}, {}] + + exception = CalendlyException(message, details) + self.assertEqual(exception.message, message) + self.assertEqual(exception.details, details) + + with self.assertRaises(CalendlyException): + raise exception + + def test_CalendlyOauth2Exception(self): + message = "SOME MESSAGE" + details = [{}, {}, {}] + + exception = CalendlyOauth2Exception(message, details) + self.assertEqual(exception.message, message) + self.assertEqual(exception.details, details) + + with self.assertRaises(CalendlyOauth2Exception): + raise exception + +class TestCalendlyReq(unittest.TestCase): + + def test_constructor(self): + pass + + def test__get_oauth2_error_from_response(self): + pass + + def test__get_api_error_from_response(self): + pass + + def test__get_error_type_and_description_from_response(self): + pass + + def test_process_request(self): + pass + + def test_get(self): + pass + + def test_post(self): + pass + + def test_delete(self): + pass + + def test_put(self): + pass + + # Set HTTP mock response class class MockResponse(object): def __init__(self, content, status_code, headers=None): @@ -28,12 +87,13 @@ def text(self): def json(self): return json.loads(self.content) + # Test endpoints class TestEndpoints(unittest.TestCase): def test_create_webhook(self): # Arrange calendly_request.post = MagicMock(return_value=MockResponse('{}', 200)) - + mock_url = 'mock_url' mock_scope = 'mock_scope' mock_organization = 'mock_organization' @@ -48,7 +108,8 @@ def test_create_webhook(self): } # Act - calendly_client.create_webhook(url=mock_url, scope=mock_scope, organization=mock_organization, signing_key=mock_signing_key) + calendly_client.create_webhook(url=mock_url, scope=mock_scope, organization=mock_organization, + signing_key=mock_signing_key) # Assert calendly_request.post.assert_called_once() @@ -57,7 +118,7 @@ def test_create_webhook(self): def test_delete_webhook(self): # Arrange calendly_request.get = MagicMock(return_value=MockResponse('{}', 204)) - + mock_uuid = 'mock_uuid' # Act @@ -71,9 +132,9 @@ def test_get_event_details(self): # Arrange with open('./calendly/tests/get_event_details_response.json', 'r') as file: calendly_request.get = MagicMock(return_value=MockResponse(file.read(), 200)) - + mock_uuid = 'mock_uuid' - + # Act response = calendly_client.get_event_details(mock_uuid) @@ -86,13 +147,13 @@ def test_list_event_types(self): # Arrange with open('./calendly/tests/list_event_types_response.json', 'r') as file: calendly_request.get = MagicMock(return_value=MockResponse(file.read(), 200)) - + mock_uuid = 'mock_uuid' expected_payload = { 'count': 20, 'user': 'mock_uuid' - } - + } + # Act response = calendly_client.list_event_types(user_uri=mock_uuid) @@ -105,13 +166,13 @@ def test_list_events(self): # Arrange with open('./calendly/tests/list_events_response.json', 'r') as file: calendly_request.get = MagicMock(return_value=MockResponse(file.read(), 200)) - + mock_uuid = 'mock_uuid' expected_payload = { 'count': 20, 'user': 'mock_uuid' - } - + } + # Act response = calendly_client.list_events(user_uri=mock_uuid) @@ -120,6 +181,7 @@ def test_list_events(self): calendly_request.get.assert_called_with(f'{constants.EVENTS}', expected_payload) self.assertEqual(response['collection'][0]['uri'], 'https://api.calendly.com/scheduled_events/MOCK_URI') + # Test endpoints class TestLogicalFunctions(unittest.TestCase): def test_get_all_items_with_pagination(self): @@ -128,12 +190,12 @@ def test_get_all_items_with_pagination(self): second_uri = 'https://api.calendly.com/event_types/B' next_page_uri = 'https://api.calendly.com/uri_to_next_page' - first_page = '{"collection": [{"uri": "'+first_uri+'"}],"pagination": {"next_page": "'+next_page_uri+'"}}' - second_page = '{"collection": [{"uri": "'+second_uri+'"}],"pagination": {"next_page": null}}' + first_page = '{"collection": [{"uri": "' + first_uri + '"}],"pagination": {"next_page": "' + next_page_uri + '"}}' + second_page = '{"collection": [{"uri": "' + second_uri + '"}],"pagination": {"next_page": null}}' calendly_client.list_event_types = MagicMock(return_value=json.loads(first_page)) calendly_client.list_events = MagicMock(return_value=json.loads(first_page)) - + # Return next page only for the correct "next_page" uri def handle_get_request(uri): if uri == next_page_uri: @@ -159,8 +221,8 @@ def test_convert_event_to_original_url_match_on_first_page(self): # Arrange with open('./calendly/tests/get_event_details_response.json', 'r') as file: calendly_client.get_event_details = MagicMock(return_value=json.loads(file.read())) - - with open('./calendly/tests/list_event_types_response.json', 'r') as file: + + with open('./calendly/tests/list_event_types_response.json', 'r') as file: calendly_client.list_event_types = MagicMock(return_value=json.loads(file.read())) mock_event_uri = 'mock_event_uri' @@ -180,8 +242,9 @@ def test_convert_event_to_original_url_paging_required(self): with open('./calendly/tests/get_event_details_response.json', 'r') as file: content = file.read() - calendly_client.get_event_details = MagicMock(side_effect=lambda x: json.loads(content) if x == mock_event_uri else None) - + calendly_client.get_event_details = MagicMock( + side_effect=lambda x: json.loads(content) if x == mock_event_uri else None) + with open('./calendly/tests/list_event_types_response.json', 'r') as file: # Result won't be found in first page mock_first_page = json.loads(file.read()) @@ -203,12 +266,244 @@ def handle_get_request(uri): calendly_request.get = MagicMock(side_effect=handle_get_request) - # Act original_url = calendly_client.convert_event_to_original_url(mock_event_uri, mock_user_uri) # Assert self.assertEqual(original_url, 'https://calendly.com/acmesales') + +class TestCalendlyOauth2(unittest.TestCase): + client_id = "ClientID123" + client_secret = "Secret123" + redirect_uri = "https://redirect.url" + response_type = "response_type" + + def test_constructor(self): + oauth2 = CalendlyOauth2(self.client_id, self.client_secret) + + self.assertEqual(oauth2.client_id, self.client_id) + self.assertEqual(oauth2.client_secret, self.client_secret) + + def test_constructor_passing_redirect_uri(self): + oauth2 = CalendlyOauth2(self.client_id, self.client_secret, redirect_uri=self.redirect_uri) + + self.assertEqual(oauth2.client_id, self.client_id) + self.assertEqual(oauth2.client_secret, self.client_secret) + self.assertEqual(oauth2.redirect_uri, self.redirect_uri) + + def test_constructor_passing_response_type(self): + oauth2 = CalendlyOauth2(self.client_id, self.client_secret) + self.assertEqual(oauth2.response_type, "code") + + oauth2 = CalendlyOauth2(self.client_id, self.client_secret, response_type=self.response_type) + self.assertEqual(oauth2.response_type, self.response_type) + + def test_authorization_url(self): + expected_url = f"{constants.OAUTH_AUTHORIZE_URL}?" \ + f"client_id={self.client_id}&response_type=code&redirect_uri={self.redirect_uri}" + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret + ) + + # redirect uri not provided + with self.assertRaises(CalendlyOauth2Exception): + _ = oauth2.authorization_url + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret, + redirect_uri=self.redirect_uri + ) + + self.assertEqual(oauth2.authorization_url, expected_url) + + def test_authorization_url_passing_response_type(self): + expected_url = f"{constants.OAUTH_AUTHORIZE_URL}?" \ + f"client_id={self.client_id}&response_type={self.response_type}&redirect_uri={self.redirect_uri}" + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret, + redirect_uri=self.redirect_uri, + response_type=self.response_type + ) + + self.assertEqual(oauth2.authorization_url, expected_url) + + @patch("calendly.utils.oauth2.CalendlyReq.post") + def test_send_post(self, post_mock): + args = (1,2,3) + kwargs = {"some": "parameters"} + response_mock = MockResponse('{}', 200) + post_mock.return_value = response_mock + + oauth2 = CalendlyOauth2(self.client_id, self.client_secret) + + returned_response = oauth2.send_post(*args, **kwargs) + self.assertEqual(response_mock, returned_response) + post_mock.assert_called_with(*args, **kwargs) + + post_mock.reset_mock() + post_mock.side_effect = CalendlyException() + + with self.assertRaises(CalendlyOauth2Exception): + oauth2.send_post(*args, **kwargs) + post_mock.assert_called_with(*args, **kwargs) + + @patch("calendly.utils.oauth2.CalendlyOauth2.send_post") + def test_get_access_token(self, send_post_mock): + code = "code123" + expected_token_data = {"TOKEN": "DATA"} + response_mock = MockResponse(json.dumps(expected_token_data), 200) + + send_post_mock.return_value = response_mock + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret + ) + + # redirect url not provided + with self.assertRaises(CalendlyOauth2Exception): + oauth2.get_access_token(code) + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret, + redirect_uri=self.redirect_uri + ) + + expected_data = dict( + grant_type="authorization_code", + client_id=self.client_id, + client_secret=self.client_secret, + code=code, + redirect_uri=self.redirect_uri + ) + + token_data = oauth2.get_access_token(code) + self.assertDictEqual(token_data, expected_token_data) + send_post_mock.assert_called_with(constants.OAUTH_TOKEN_URL, expected_data) + + @patch("calendly.utils.oauth2.CalendlyOauth2.send_post") + def test_get_access_token_passing_grant_type(self, send_post_mock): + code = "code123" + grant_type = "TYPE" + expected_token_data = {"TOKEN": "DATA"} + response_mock = MockResponse(json.dumps(expected_token_data), 200) + + send_post_mock.return_value = response_mock + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret, + redirect_uri=self.redirect_uri + ) + + expected_data = dict( + grant_type=grant_type, + client_id=self.client_id, + client_secret=self.client_secret, + code=code, + redirect_uri=self.redirect_uri + ) + + token_data = oauth2.get_access_token(code, grant_type) + self.assertDictEqual(token_data, expected_token_data) + send_post_mock.assert_called_with(constants.OAUTH_TOKEN_URL, expected_data) + + @patch("calendly.utils.oauth2.CalendlyOauth2.send_post") + def test_revoke_access_token(self, send_post_mock): + token = "TOKEN123" + expected_token_data = {"TOKEN": "DATA"} + response_mock = MockResponse(json.dumps(expected_token_data), 200) + expected_data = dict(client_id=self.client_id, client_secret=self.client_secret, token=token) + send_post_mock.return_value = response_mock + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret, + redirect_uri=self.redirect_uri + ) + + oauth2.revoke_access_token(token) + send_post_mock.assert_called_with(constants.OAUTH_REVOKE_URL, expected_data) + + @patch("calendly.utils.oauth2.CalendlyOauth2.send_post") + def test_refresh_access_token(self, send_post_mock): + refresh_token = "TOKEN123" + expected_token_data = {"TOKEN": "DATA"} + response_mock = MockResponse(json.dumps(expected_token_data), 200) + + expected_data = dict( + grant_type="refresh_token", + client_id=self.client_id, + client_secret=self.client_secret, + refresh_token=refresh_token + ) + + send_post_mock.return_value = response_mock + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret, + redirect_uri=self.redirect_uri + ) + + oauth2.refresh_access_token(refresh_token) + send_post_mock.assert_called_with(constants.OAUTH_TOKEN_URL, expected_data) + + @patch("calendly.utils.oauth2.CalendlyOauth2.send_post") + def test_refresh_access_token_passing_grant_type(self, send_post_mock): + grant_type = "grant_type" + refresh_token = "TOKEN123" + expected_token_data = {"TOKEN": "DATA"} + response_mock = MockResponse(json.dumps(expected_token_data), 200) + + expected_data = dict( + grant_type=grant_type, + client_id=self.client_id, + client_secret=self.client_secret, + refresh_token=refresh_token + ) + + send_post_mock.return_value = response_mock + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret, + redirect_uri=self.redirect_uri + ) + + oauth2.refresh_access_token(refresh_token, grant_type) + send_post_mock.assert_called_with(constants.OAUTH_TOKEN_URL, expected_data) + + @patch("calendly.utils.oauth2.CalendlyOauth2.send_post") + def test_introspect_access_token(self, send_post_mock): + token = "TOKEN123" + expected_token_data = {"TOKEN": "DATA"} + response_mock = MockResponse(json.dumps(expected_token_data), 200) + + expected_data = dict( + client_id=self.client_id, + client_secret=self.client_secret, + token=token + ) + + send_post_mock.return_value = response_mock + + oauth2 = CalendlyOauth2( + self.client_id, + self.client_secret, + redirect_uri=self.redirect_uri + ) + + oauth2.introspect_access_token(token) + send_post_mock.assert_called_with(constants.OAUTH_INTROSPECT_URL, expected_data) + + if __name__ == '__main__': - unittest.main() \ No newline at end of file + unittest.main() diff --git a/calendly/utils/requests.py b/calendly/utils/api.py similarity index 52% rename from calendly/utils/requests.py rename to calendly/utils/api.py index e468819..9ad1a97 100644 --- a/calendly/utils/requests.py +++ b/calendly/utils/api.py @@ -1,11 +1,10 @@ +from typing import MutableMapping +from calendly.exceptions import CalendlyException import requests -from typing import List, MutableMapping __author__ = "laxmena " __license__ = "MIT" -class CalendlyException(Exception): - """Errors corresponding to a misuse of Calendly API""" class CalendlyReq(object): """ @@ -16,16 +15,64 @@ class CalendlyReq(object): https://calendly.stoplight.io/docs/api-docs/ """ - def __init__(self, token: str): + OAUTH2_ERROR_TYPE_KEY = 'error' + OAUTH2_ERROR_DESCRIPTION_KEY = 'error_description' + + API_ERROR_TYPE_KEY = "title" + API_ERROR_DESCRIPTION_KEY = "message" + API_ERROR_DETAILS_KEY = "details" + + def __init__(self, token: str=None, headers: dict=None): """ - Constructor. Uses Bearer Token Authentication. + Constructor: Uses Bearer Token Authentication or custom headers. Parameters ---------- token : str + headers : str Personal Access Token """ - self.headers = {'authorization': 'Bearer ' + token} + + if token and headers: + raise CalendlyException("You can't pass both token and headers at the same time.") + + if token: + headers = {'authorization': 'Bearer ' + token} + + self.headers = headers + + def _get_oauth2_error_from_response(self,response): + try: + resp = response.json() + return resp[self.OAUTH2_ERROR_TYPE_KEY], resp[self.OAUTH2_ERROR_DESCRIPTION_KEY], [] + except (AttributeError, KeyError): + return + + def _get_api_error_from_response(self, response): + + try: + resp = response.json() + errors = [resp[self.API_ERROR_TYPE_KEY], resp[self.API_ERROR_DESCRIPTION_KEY]] + except (AttributeError, KeyError): + return + + try: + errors.append(resp[self.API_ERROR_DETAILS_KEY]) + except (AttributeError, KeyError): + errors.append([]) + + return tuple(errors) + + def _get_error_type_and_description_from_response(self, response): + + oauth2_errors = self._get_oauth2_error_from_response(response) + if not oauth2_errors: + oauth2_errors = self._get_api_error_from_response(response) + + if not oauth2_errors: + oauth2_errors = "error", "Unknown Error.", [] + + return oauth2_errors def process_request(self, method: str, url: str, data: MutableMapping=None) -> requests.Response: """ @@ -41,7 +88,17 @@ def process_request(self, method: str, url: str, data: MutableMapping=None) -> r additional data to be passed to the API """ request_method = getattr(requests, method) - return request_method(url, json=data, headers=self.headers) + kwargs = dict(json=data) + + if self.headers: + kwargs.update(dict(headers=self.headers)) + + response = request_method(url, **kwargs) + if response.status_code > requests.codes.permanent_redirect: + error_type, error_description, error_details = self._get_error_type_and_description_from_response(response) + raise CalendlyException(f"{error_type}: {error_description}", error_details) + + return response def get(self, url: str, data: MutableMapping=None) -> requests.Response: """ diff --git a/calendly/utils/constants.py b/calendly/utils/constants.py index fb3e098..a931dae 100644 --- a/calendly/utils/constants.py +++ b/calendly/utils/constants.py @@ -1,3 +1,4 @@ + BASE="https://api.calendly.com" WEBHOOK=f"{BASE}/webhook_subscriptions" USERS=f"{BASE}/users" @@ -9,3 +10,10 @@ ORGANIZATIONS=f"{BASE}/organizations/" ORGANIZATION_MEMBERSHIPS=f"{BASE}/organization_memberships" DATA_COMPLIANCE=f"{BASE}/data_compliance/deletion/invitees" + +OAUTH_BASE_URL = "https://auth.calendly.com/oauth" +OAUTH_AUTHORIZE_URL = f"{OAUTH_BASE_URL}/authorize" +OAUTH_TOKEN_URL = f"{OAUTH_BASE_URL}/token" +OAUTH_REVOKE_URL = f"{OAUTH_BASE_URL}/revoke" +OAUTH_INTROSPECT_URL = f"{OAUTH_BASE_URL}/introspect" + diff --git a/calendly/utils/oauth2.py b/calendly/utils/oauth2.py new file mode 100644 index 0000000..af7f756 --- /dev/null +++ b/calendly/utils/oauth2.py @@ -0,0 +1,102 @@ +from calendly.exceptions import CalendlyOauth2Exception, CalendlyException +from .api import CalendlyReq +from .constants import OAUTH_AUTHORIZE_URL, OAUTH_TOKEN_URL, OAUTH_REVOKE_URL, OAUTH_INTROSPECT_URL + +__author__ = "luis " +__license__ = "MIT" + +REDIRECT_URI_EXCEPTION_TEXT = "You must pass the redirect_uri in the CalendlyOauth2 instantiation." + +class CalendlyOauth2(object): + """ + Private class wrapping the Calendly Oauth2 Api + + References + ---------- + https://developer.calendly.com/how-to-authenticate-with-oauth + + """ + + client_id = None + client_secret = None + redirect_uri = None + response_type = None + + def __init__(self, client_id: str, client_secret: str, redirect_uri: str=None, response_type: str=None): + """ + Constructor. client_id, client_secret, optionally you can pass the redirect_uri and response type + + Parameters + ---------- + client_id : str + client_secret : str + redirect_uri : str + response_type : str + """ + self.request = CalendlyReq() + self.client_id = client_id + self.client_secret = client_secret + self.redirect_uri = redirect_uri + self.response_type = response_type or "code" + + @property + def authorization_url(self): + if not self.redirect_uri: + raise CalendlyOauth2Exception(REDIRECT_URI_EXCEPTION_TEXT) + + return f"{OAUTH_AUTHORIZE_URL}?client_id={self.client_id}&response_type={self.response_type}&redirect_uri={self.redirect_uri}" + + def send_post(self, *args, **kwargs): + try: + return self.request.post(*args, **kwargs) + except CalendlyException as e: + raise CalendlyOauth2Exception(str(e)) + + def get_access_token(self, code: str, grant_type: str=None): + + if not self.redirect_uri: + raise CalendlyOauth2Exception(REDIRECT_URI_EXCEPTION_TEXT) + + data = dict( + grant_type=grant_type or "authorization_code", + client_id=self.client_id, + client_secret=self.client_secret, + code=code, + redirect_uri=self.redirect_uri + ) + + response = self.send_post(OAUTH_TOKEN_URL, data) + return response.json() + + def revoke_access_token(self, token: str): + data = dict( + client_id=self.client_id, + client_secret=self.client_secret, + token=token + ) + + response = self.send_post(OAUTH_REVOKE_URL, data) + return response.json() + + def refresh_access_token(self, refresh_token: str, grant_type: str=None): + data = dict( + grant_type=grant_type or "refresh_token", + client_id=self.client_id, + client_secret=self.client_secret, + refresh_token=refresh_token + ) + + response = self.send_post(OAUTH_TOKEN_URL, data) + return response.json() + + def introspect_access_token(self, token: str): + data = dict( + client_id=self.client_id, + client_secret=self.client_secret, + token=token + ) + + response = self.send_post(OAUTH_INTROSPECT_URL, data) + return response.json() + +