-
Notifications
You must be signed in to change notification settings - Fork 2
/
gateway.py
145 lines (124 loc) · 6.37 KB
/
gateway.py
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
142
143
144
145
"""
Gateway
=======
This module is the primary class for communicating with the AkuLaku API.
"""
import base64
from datetime import datetime
import hashlib
import hmac
import json
import logging
from http import HTTPStatus
import requests
from ovoclient.exceptions import OvoClientError
from ovoclient.models import PaymentRequest, PaymentResponse, ResponseCode
log = logging.getLogger()
__all__ = ['OvoClientGateway', ]
CHARGE_DEFAULT_TIMEOUT = 65
REVERSAL_DEFAULT_TIMEOUT = 15
MAX_REVERSAL_COUNT = 3
class OvoClientGateway:
def __init__(self, app_id, secret_key, use_sandbox=False):
self.app_id = app_id
self.secret_key = secret_key
self.use_sandbox = use_sandbox
@property
def base_url(self):
""" Get base URL for the gateway
"""
return "https://api.byte-stack.net" if self.use_sandbox \
else "https://api.ovo.id"
def generate_signature(self, random: int):
return hmac.new(self.secret_key.encode(),
f'{self.app_id}{random}'.encode(), hashlib.sha256).hexdigest()
def charge(self, payment_request: PaymentRequest, timeout: int = None):
"""Send push to pay transaction (CHARGE)
:param `ovoclient.models.PaymentRequest payment_request:
:param timeout: int, optional, how long we took time before deciding connection timeout
:return:
"""
data = payment_request.serialize()
random_number = int(datetime.now().timestamp())
headers = {
'app-id': self.app_id,
'Random': f"{random_number}",
'Hmac': self.generate_signature(random_number),
'Content-Type': 'application/json'
}
log.info(f"Processing ovo payment {payment_request.reference_number}")
log.info(f"OVO transaction {datetime.now().isoformat()}: {data}")
try:
url = f'{self.base_url}/pos'
response = requests.post(url=url,
data=json.dumps(data),
headers=headers,
timeout=CHARGE_DEFAULT_TIMEOUT if not timeout else timeout)
response_json = json.loads(response.content.decode('utf-8')) if response.content else {}
if response.status_code == HTTPStatus.NOT_FOUND and not response_json:
response_json = data
response_json['responseCode'] = ResponseCode.NOT_FOUND.value
log.exception(f"Failed to create new ovo payment for order {payment_request.reference_number}")
if response.status_code not in [HTTPStatus.OK, HTTPStatus.NOT_FOUND] and not response_json:
raise OvoClientError("Failed to register into ovo api")
except (requests.ConnectTimeout, requests.HTTPError, requests.ConnectionError):
response_json = data
log.exception(f"Failed to create new ovo payment for order {payment_request.reference_number}")
transaction_request_data = data.get('transactionRequestData', {})
response_json['transactionRequestData']['phone'] = transaction_request_data.get('phone', '')
response_json['transactionRequestData']['merchantInvoice'] = transaction_request_data.get('merchantInvoice',
'')
return PaymentResponse.from_api_json(response_json)
def reversal(self, payment_request: PaymentRequest, timeout: int = None, attempt: int = 0):
"""Send push to pay transaction (REVERSE)
:param `ovoclient.models.PaymentRequest payment_request:
:return:
"""
data = payment_request.serialize()
random_number = int(datetime.now().timestamp())
headers = {
'app-id': self.app_id,
'Random': f"{random_number}",
'Hmac': self.generate_signature(random_number),
'Content-Type': 'application/json'
}
log.info(f"Processing ovo reversal {payment_request.reference_number}")
log.info(f"OVO reversal transaction {datetime.now().isoformat()}: {data}")
try:
url = f'{self.base_url}/pos'
attempt += 1
response = requests.post(url=url,
data=json.dumps(data),
headers=headers,
timeout=REVERSAL_DEFAULT_TIMEOUT if not timeout else timeout)
response_json = json.loads(response.content.decode('utf-8')) if response.content else {}
if response.status_code == HTTPStatus.NOT_FOUND and not response_json:
response_json = data
response_json['responseCode'] = ResponseCode.NOT_FOUND.value
log.exception(f"Failed to create reversal for order {payment_request.reference_number}")
if response.status_code != HTTPStatus.OK and not response_json:
raise OvoClientError("Failed to register into ovo api")
except (requests.ConnectTimeout, requests.HTTPError, requests.ConnectionError, requests.ReadTimeout):
response_json = data
log.exception(f"Failed to create new ovo reversal for order {payment_request.reference_number}")
transaction_request_data = data.get('transactionRequestData', {})
response_json['transactionRequestData']['phone'] = transaction_request_data.get('phone', '')
response_json['transactionRequestData']['merchantInvoice'] = transaction_request_data.get('merchantInvoice',
'')
return PaymentResponse.from_api_json(response_json), attempt
def recursive_reversal(self, payment_request: PaymentRequest, timeout: int = None):
"""
Recursive reversal based on MAX_REVERSAL_COUNT and response from reversal whether its succeeded
:param payment_request:
:param timeout:
:return:
"""
attempt = 0
response_json = None
while attempt < MAX_REVERSAL_COUNT and (
not response_json or
response_json.response_code != ResponseCode.SUCCESS.value
):
payment_request.date = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')
response_json, attempt = self.reversal(payment_request=payment_request, timeout=timeout, attempt=attempt)
return response_json