-
Notifications
You must be signed in to change notification settings - Fork 0
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Bamboo sdk base api class #97
Changes from 4 commits
e7dfcb3
0e63fc6
b5b5943
e2e45dd
0427fac
a9b6ebd
6788d58
5867d4b
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,72 @@ | ||
import requests | ||
import base64 | ||
import json | ||
from bamboosdk.exceptions import * | ||
|
||
|
||
payload = { "fields": ["displayName", "firstName", "lastName", "department", "workEmail", "supervisorEmail", "status"] } | ||
|
||
class ApiBase: | ||
""" | ||
Base class for all API classes | ||
""" | ||
API_BASE_URL = 'https://api.bamboohr.com/api/gateway.php/{}' | ||
|
||
def __init__(self) -> None: | ||
self.__api_token = None | ||
self.__sub_domain = None | ||
self.headers = None | ||
|
||
def _post_request(self, module_api_path): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. we'll need a new func for generator too |
||
""" | ||
HTTP post method to send data to BambooHR API URL | ||
|
||
Parameters: | ||
payload (dict): Data to be sent to Bamboo API | ||
module_api_path (str): URL of BambooHR API | ||
""" | ||
|
||
url= self.API_BASE_URL.format(self.__sub_domain) + module_api_path | ||
response = requests.post(url=url, json=payload, headers=self.headers) | ||
if response.status_code == 200: | ||
result = json.loads(response.text) | ||
return result | ||
|
||
if response.status_code == 403: | ||
error_msg = json.loads(response.text) | ||
raise NoPrivilegeError('Forbidden, the user has insufficient privilege', error_msg) | ||
|
||
if response.status_code == 404: | ||
error_msg = json.loads(response.text) | ||
raise NotFoundItemError('Not found item with ID', error_msg) | ||
|
||
if response.status_code == 401: | ||
error_msg = 'The api token is invalid' | ||
raise InvalidTokenError('Invalid token, try to refresh it', error_msg) | ||
Comment on lines
+96
to
+106
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. checkout docs once and see if these status codes are expected, also in the end add raise 1 exception to handle status codes that are not handled above, 5xx for example There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. i have checked the docs these are the errors there |
||
|
||
else: | ||
error_msg = 'Something went wrong' | ||
raise Exception('Something went wrong') | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. send response.text to the exception, also create 1 exception, something like BambooHRSDKError() instead of generic one |
||
|
||
|
||
def __encode_username_password(self): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why are we encoding and decoding it again There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the method b64encode converts the string to base64 byte code which doesn't work when sending in the header and the encoding it done on the this 'byte' its only making that to base64 string. |
||
""" | ||
Utility method to be used in the header for authorization | ||
converts the api token and password to base64 | ||
""" | ||
|
||
payload = f'{self.__api_token}:a' | ||
payload = payload.encode() | ||
|
||
return base64.b64encode(payload).decode() | ||
|
||
def set_api_token(self, api_token): | ||
self.__api_token = api_token | ||
|
||
self.headers = { | ||
"content-type": "application/json", | ||
"authorization": f"Basic {self.__encode_username_password()}" | ||
} | ||
|
||
def set_sub_domain(self, sub_domain): | ||
self.__sub_domain = sub_domain |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,27 @@ | ||
""" | ||
BambooHR SDK exceptions | ||
""" | ||
|
||
|
||
class BambooHrSDKError(Exception): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @Ashutosh619-sudo make sure this contains all the errors |
||
""" | ||
The base exception class for BambooHR SDK | ||
""" | ||
|
||
def __init__(self, msg, response=None): | ||
super(BambooHrSDKError, self).__init__(msg) | ||
self.message = msg | ||
self.response = response | ||
|
||
def __str__(self): | ||
return repr(self.message) | ||
|
||
|
||
class NoPrivilegeError(BambooHrSDKError): | ||
"""The user has insufficient privilege, 403 error.""" | ||
|
||
class NotFoundItemError(BambooHrSDKError): | ||
"""Not found the item from URL, 404 error.""" | ||
|
||
class InvalidTokenError(BambooHrSDKError): | ||
"""Invalid or non-existing access token, 401 error""" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@Ashutosh619-sudo we can create a new folders called connectors
and moved this inside that folder and name it bamboohr
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this is fine for now, connectors sounds very generic. We'll just assume we've kept the SDK folder inside API, that's it.