-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathapi_client.py
More file actions
141 lines (109 loc) · 6.55 KB
/
Copy pathapi_client.py
File metadata and controls
141 lines (109 loc) · 6.55 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
import base64
import hashlib
import hmac
import json
import time
from datetime import datetime
from typing import Generator
import requests
from solidgate.encryption import AESCipher
from solidgate.model import FormInitDTO, FormUpdateDTO, FormResignDTO
class ApiClient:
BASE_SOLID_GATE_API_URI = 'https://pay.solidgate.com/api/v1/'
BASE_RECONCILIATION_API_URI = 'https://reports.solidgate.com/'
RECONCILIATION_ORDERS_PATH = 'api/v2/reconciliation/orders'
RECONCILIATION_CHARGEBACKS_PATH = 'api/v2/reconciliation/chargebacks'
RECONCILIATION_DATETIME_FORMAT = '%Y-%m-%d %H:%M:%S'
DEFAULT_RETRY_DELAY = 5
MAX_ATTEMPTS = 5
RETRYABLE_STATUS_CODES = (429, 503)
USER_AGENT = 'Python-SDK-0.4.0'
def __init__(self, merchant_id: str,
private_key: str,
base_solid_gate_api_uri: str = BASE_SOLID_GATE_API_URI,
base_reconciliations_api_uri: str = BASE_RECONCILIATION_API_URI):
self.__merchant_id = merchant_id
self.__private_key = private_key
self.__base_solid_gate_api_uri = base_solid_gate_api_uri
self.__base_reconciliations_api_uri = base_reconciliations_api_uri
def charge(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('charge', attributes)
def recurring(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('recurring', attributes)
def resign(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('resign', attributes)
def status(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('status', attributes)
def refund(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('refund', attributes)
def auth(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('auth', attributes)
def void(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('void', attributes)
def settle(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('settle', attributes)
def arn_code(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('arn-code', attributes)
def apple_pay(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('apple-pay', attributes)
def google_pay(self, attributes: dict) -> requests.models.Response:
return self.__send_solidgate_request('google-pay', attributes)
def form_merchant_data(self, attributes: dict) -> FormInitDTO:
payment_intent = AESCipher(self.__private_key).encrypt(self.__convert_request_attributes_to_str(attributes))
signature = self.__generate_signature(payment_intent)
form_init_dto = FormInitDTO(payment_intent=payment_intent, publicKey=self.__merchant_id, signature=signature)
return form_init_dto
def form_update(self, attributes: dict) -> FormUpdateDTO:
partial_intent = AESCipher(self.__private_key).encrypt(self.__convert_request_attributes_to_str(attributes))
signature = self.__generate_signature(partial_intent)
form_update_dto = FormUpdateDTO(partial_intent=partial_intent, signature=signature)
return form_update_dto
def form_resign(self, attributes: dict) -> FormInitDTO:
resign_intent = AESCipher(self.__private_key).encrypt(self.__convert_request_attributes_to_str(attributes))
signature = self.__generate_signature(resign_intent)
form_resign_dto = FormResignDTO(resign_intent=resign_intent, publicKey=self.__merchant_id, signature=signature)
return form_resign_dto
def order_reconciliation(self, date_from: datetime, date_to: datetime) -> Generator:
return self.__send_reconciliation_request(self.RECONCILIATION_ORDERS_PATH, date_from, date_to)
def chargeback_reconciliation(self, date_from: datetime, date_to: datetime) -> Generator:
return self.__send_reconciliation_request(self.RECONCILIATION_CHARGEBACKS_PATH, date_from, date_to)
def __generate_signature(self, data: str) -> str:
encrypto_data = (self.__merchant_id + data + self.__merchant_id).encode('utf-8')
sign = hmac.new(self.__private_key.encode('utf-8'), encrypto_data, hashlib.sha512).hexdigest()
return base64.b64encode(sign.encode('utf-8')).decode('utf-8')
def __send_solidgate_request(self, method: str, attributes: dict) -> requests.models.Response:
full_url = self.__base_solid_gate_api_uri + method
return self.__send_request(full_url, attributes)
def __send_reconciliation_request(self, path: str, date_from: datetime, date_to: datetime) -> Generator:
full_url = self.__base_reconciliations_api_uri + path
attributes = {'date_from': date_from.strftime(self.RECONCILIATION_DATETIME_FORMAT),
'date_to': date_to.strftime(self.RECONCILIATION_DATETIME_FORMAT)}
while True:
try:
response = self.__send_request(full_url, attributes)
except requests.exceptions.RequestException as err:
raise Exception("Exception during request:", err)
if 200 <= response.status_code < 300:
response_content = json.loads(response.content)
for order in response_content['orders']:
yield order
attributes['next_page_iterator'] = response_content['metadata']['next_page_iterator']
if not attributes['next_page_iterator']:
break
else:
raise Exception('Request is not successful!')
def __send_request(self, path: str, attributes: dict) -> requests.models.Response:
body = self.__convert_request_attributes_to_str(attributes)
headers = {'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': self.USER_AGENT,
'Merchant': self.__merchant_id,
'Signature': self.__generate_signature(body)}
for attempt in range(1, self.MAX_ATTEMPTS + 1):
response = requests.post(path, headers=headers, json=attributes)
if response.status_code not in self.RETRYABLE_STATUS_CODES or attempt == self.MAX_ATTEMPTS:
return response
time.sleep(self.DEFAULT_RETRY_DELAY)
@staticmethod
def __convert_request_attributes_to_str(attributes: dict) -> str:
return json.dumps(attributes)