XF/pluto
XF
/
pluto
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Port Agora and LBTC API libraries to use treq

This commit is contained in:
Mark Veidemanis 2022-05-02 18:20:32 +01:00
parent 1f0dc33250
commit fba0aab966
Signed by: m
GPG Key ID: 5ACFCEED46C0904F
2 changed files with 121 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
handler/lib/agoradesk_py.py
View File

@ -10,7 +10,8 @@ from typing import Optional
from typing import Union
import arrow
import httpx
import treq
from twisted.internet.defer import inlineCallbacks
# Project imports
import util
@ -54,6 +55,25 @@ class AgoraDesk:
logger.debug("creating instance of AgoraDesk API with api_key %s", self.api_key)
@inlineCallbacks
def callback_api_call(self, response, result):
logger.debug(response)
text = yield response.content()
try:
result["response"] = json.loads(text)
except json.decoder.JSONDecodeError:
result["success"] = "ERROR"
result["message"] = "Error parsing JSON."
return result
result["status"] = response.code
if response.code == 200:
result["success"] = True
result["message"] = "OK"
else:
result["message"] = "API ERROR"
return result
def _api_call(
self,
api_method: str,
@ -81,51 +101,53 @@ class AgoraDesk:
"status": None,
}
try:
response = None
if http_method == "POST":
if query_values:
response = httpx.post(
url=api_call_url,
headers=headers,
content=json.dumps(query_values),
)
else:
response = httpx.post(
url=api_call_url,
headers=headers,
)
response = None
if http_method == "POST":
if query_values:
# response = httpx.post(
# url=api_call_url,
# headers=headers,
# content=json.dumps(query_values),
# )
response = treq.post(
api_call_url,
headers=headers,
data=json.dumps(query_values),
)
else:
response = httpx.get(url=api_call_url, headers=headers, params=query_values)
# response = httpx.post(
# url=api_call_url,
# headers=headers,
# )
response = treq.post(
api_call_url,
headers=headers,
)
logger.debug(response)
result["response"] = json.loads(response.text)
result["status"] = response.status_code
if response.status_code == 200:
result["success"] = True
result["message"] = "OK"
else:
result["message"] = "API ERROR"
else:
# response = httpx.get(url=api_call_url, headers=headers, params=query_values)
response = treq.get(api_call_url, headers=headers, params=query_values)
if response:
response.addCallback(self.callback_api_call, result)
return response
return result
except httpx.ConnectError as error:
result["message"] = str(error)
result["status"] = 600
result["response"] = {"error": {"message": error}}
return result
except json.decoder.JSONDecodeError:
result["message"] = "Not JSON"
if response:
result["status"] = response.status_code
result["response"] = {"error": {"message": response.text}}
return result
except httpx.ReadTimeout:
result["message"] = "Read timed out"
if response:
result["status"] = response.status_code
result["response"] = {"error": {"message": response.text}}
return result
# except httpx.ConnectError as error:
# result["message"] = str(error)
# result["status"] = 600
# result["response"] = {"error": {"message": error}}
# return result
# except json.decoder.JSONDecodeError:
# result["message"] = "Not JSON"
# if response:
# result["status"] = response.status_code
# result["response"] = {"error": {"message": response.text}}
# return result
# except httpx.ReadTimeout:
# result["message"] = "Read timed out"
# if response:
# result["status"] = response.status_code
# result["response"] = {"error": {"message": response.text}}
# return result
# Account related API Methods
# ===========================

56
handler/lib/localbitcoins_py.py
View File

@ -18,6 +18,8 @@ import hmac as hmac_lib
import requests
import sys
import time
import treq
from twisted.internet.defer import inlineCallbacks
from urllib.parse import urlparse
@ -98,6 +100,24 @@ class LocalBitcoins:
params_encoded = urlparse(api_request.url).query
return (api_request, params_encoded)
@inlineCallbacks
def callback_api_call(self, response, result):
logger.debug(response)
text = yield response.content()
try:
result["response"] = json.loads(text)
except json.decoder.JSONDecodeError:
result["success"] = "ERROR"
result["message"] = "Error parsing JSON."
return result
result["status"] = response.code
if response.code == 200:
result["success"] = True
result["message"] = "OK"
else:
result["message"] = "API ERROR"
return result
def _api_call(
self,
api_method: str,
@ -116,6 +136,12 @@ class LocalBitcoins:
nonce = str(int(time.time() * 1000)).encode("ascii")
signature = self.sign_payload(nonce, url, params_encoded)
headers = {
"Apiauth-Key": self.hmac_key,
"Apiauth-Nonce": nonce,
"Apiauth-Signature": signature,
}
api_request.headers["Apiauth-Key"] = self.hmac_key
api_request.headers["Apiauth-Nonce"] = nonce
api_request.headers["Apiauth-Signature"] = signature
@ -132,6 +158,36 @@ class LocalBitcoins:
"response": None,
"status": None,
}
response = None
if http_method == "POST":
if query_values:
# response = httpx.post(
# url=api_call_url,
# headers=headers,
# content=json.dumps(query_values),
# )
response = treq.post(
api_call_url,
headers=headers,
data=json.dumps(query_values),
)
else:
# response = httpx.post(
# url=api_call_url,
# headers=headers,
# )
response = treq.post(
api_call_url,
headers=headers,
)
else:
# response = httpx.get(url=api_call_url, headers=headers, params=query_values)
response = treq.get(api_call_url, headers=headers, params=query_values)
if response:
response.addCallback(self.callback_api_call, result)
return response
session = requests.Session()
try:
response = session.send(api_request)