Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def mobilePaymentB2CRequest(self, productName_, recipients_):
parameters = {
'username' : self.username,
'productName' : productName_,
'recipients' : recipients_
}
url = self.getMobilePaymentB2CUrl()
response = self.sendJSONRequest(url, json.dumps(parameters))
if self.responseCode == self.HTTP_RESPONSE_CREATED:
decoded = json.loads(response)
if len(decoded['entries']) > 0:
return decoded['entries']
raise AfricasTalkingGatewayException(decoded['errorMessage'])
raise AfricasTalkingGatewayException(response)
def deleteSubscription(self, phoneNumber_, shortCode_, keyword_):
if len(phoneNumber_) == 0 or len(shortCode_) == 0 or len(keyword_) == 0:
raise AfricasTalkingGatewayException("Please supply phone number, short code and keyword")
url = "%s/delete" %(self.getSmsSubscriptionUrl())
parameters = {
'username' :self.username,
'phoneNumber' :phoneNumber_,
'shortCode' :shortCode_,
'keyword' :keyword_
}
response = self.sendRequest(url, parameters)
if self.responseCode == self.HTTP_RESPONSE_CREATED:
decoded = json.loads(response)
return decoded
raise AfricasTalkingGatewayException(response)
def uploadMediaFile(self, urlString_):
parameters = {
'username' :self.username,
'url' :urlString_
}
url = "%s/mediaUpload" %(self.getVoiceUrl())
response = self.sendRequest(url, parameters)
decoded = json.loads(response)
if decoded['errorMessage'] != "None":
raise AfricasTalkingGatewayException(decoded['errorMessage'])
def sendAirtime(self, recipients_):
parameters = {
'username' : self.username,
'recipients' : json.dumps(recipients_)
}
url = "%s/send" %(self.getAirtimeUrl())
response = self.sendRequest(url, parameters)
decoded = json.loads(response)
responses = decoded['responses']
if self.responseCode == self.HTTP_RESPONSE_CREATED:
if len(responses) > 0:
return responses
raise AfricasTalkingGatewayException(decoded["errorMessage"])
raise AfricasTalkingGatewayException(response)
def sendJSONRequest(self, urlString, data_):
try:
headers = { 'Accept' : 'application/json',
'Content-Type' : 'application/json',
'apikey' : self.apiKey }
resp = requests.post(urlString, data = data_, headers = headers)
except requests.exceptions.Timeout as e:
raise AfricasTalkingGatewayException(e.read())
except requests.exceptions.RequestException as e:
raise AfricasTalkingGatewayException(e.read())
else:
self.responseCode = resp.status_code
response = resp.text
if self.Debug:
print("Raw response: " + response)
return response
def sendAirtime(self, recipients_):
parameters = {
'username' : self.username,
'recipients' : json.dumps(recipients_)
}
url = "%s/send" %(self.getAirtimeUrl())
response = self.sendRequest(url, parameters)
decoded = json.loads(response)
responses = decoded['responses']
if self.responseCode == self.HTTP_RESPONSE_CREATED:
if len(responses) > 0:
return responses
raise AfricasTalkingGatewayException(decoded["errorMessage"])
raise AfricasTalkingGatewayException(response)
parameters = {
'username' : self.username,
'productName' : productName_,
'phoneNumber' : phoneNumber_,
'currencyCode' : currencyCode_,
'providerChannel' : providerChannel_,
'amount' : amount_,
'metadata' : metadata_
}
url = self.getMobilePaymentCheckoutUrl()
response = self.sendJSONRequest(url, json.dumps(parameters))
if self.responseCode == self.HTTP_RESPONSE_CREATED:
decoded = json.loads(response)
if decoded['status'] == 'PendingConfirmation':
return decoded['transactionId']
raise AfricasTalkingGatewayException(decoded['description'])
raise AfricasTalkingGatewayException(response)
'username' : self.username,
'productName' : productName_,
'phoneNumber' : phoneNumber_,
'currencyCode' : currencyCode_,
'providerChannel' : providerChannel_,
'amount' : amount_,
'metadata' : metadata_
}
url = self.getMobilePaymentCheckoutUrl()
response = self.sendJSONRequest(url, json.dumps(parameters))
if self.responseCode == self.HTTP_RESPONSE_CREATED:
decoded = json.loads(response)
if decoded['status'] == 'PendingConfirmation':
return decoded['transactionId']
raise AfricasTalkingGatewayException(decoded['description'])
raise AfricasTalkingGatewayException(response)
def deleteSubscription(self, phoneNumber_, shortCode_, keyword_):
if len(phoneNumber_) == 0 or len(shortCode_) == 0 or len(keyword_) == 0:
raise AfricasTalkingGatewayException("Please supply phone number, short code and keyword")
url = "%s/delete" %(self.getSmsSubscriptionUrl())
parameters = {
'username' :self.username,
'phoneNumber' :phoneNumber_,
'shortCode' :shortCode_,
'keyword' :keyword_
}
response = self.sendRequest(url, parameters)
if self.responseCode == self.HTTP_RESPONSE_CREATED:
decoded = json.loads(response)
return decoded
raise AfricasTalkingGatewayException(response)