aljaz/bitfinex-api-py
1
0
Fork 0
mirror of https://github.com/aljazceru/bitfinex-api-py.git synced 2025-12-20 15:24:21 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add new module bfxapi._utils.json_decoder.

Browse Source
This commit is contained in:
Davide Casale
2023-10-08 21:49:36 +02:00
parent 9872adf60f
commit de0ee54900
6 changed files with 107 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
bfxapi/rest/endpoints/rest_merchant_endpoints.py
View File

@@ -1,46 +1,32 @@
import re
from typing import Callable, TypeVar, cast, \
TypedDict, Dict, List, Union, Literal, Optional, Any
from typing import \
TypedDict, Dict, List, \
Union, Literal, Optional, \
Any
from decimal import Decimal
from ..middleware import Middleware
from bfxapi.rest.middleware import Middleware
from ..enums import MerchantSettingsKey
from bfxapi.rest.enums import MerchantSettingsKey
from ...types import \
InvoiceSubmission, InvoicePage, InvoiceStats, \
CurrencyConversion, MerchantDeposit, MerchantUnlinkedDeposit
#region Defining methods to convert dictionary keys to snake_case and camelCase.
T = TypeVar("T")
_to_snake_case: Callable[[str], str] = lambda string: re.sub(r"(?<!^)(?=[A-Z])", "_", string).lower()
_to_camel_case: Callable[[str], str] = lambda string: \
(components := string.split("_"))[0] + str().join(c.title() for c in components[1:])
def _scheme(data: T, adapter: Callable[[str], str]) -> T:
if isinstance(data, list):
return cast(T, [ _scheme(sub_data, adapter) for sub_data in data ])
if isinstance(data, dict):
return cast(T, { adapter(key): _scheme(value, adapter) for key, value in data.items() })
return data
def _to_snake_case_keys(dictionary: T) -> T:
return _scheme(dictionary, _to_snake_case)
def _to_camel_case_keys(dictionary: T) -> T:
return _scheme(dictionary, _to_camel_case)
#endregion
from bfxapi.types import \
InvoiceSubmission, \
InvoicePage, \
InvoiceStats, \
CurrencyConversion, \
MerchantDeposit, \
MerchantUnlinkedDeposit
_CustomerInfo = TypedDict("_CustomerInfo", {
"nationality": str, "resid_country": str, "resid_city": str,
"resid_zip_code": str, "resid_street": str, "resid_building_no": str,
"full_name": str, "email": str, "tos_accepted": bool
"nationality": str,
"resid_country": str,
"resid_city": str,
"resid_zip_code": str,
"resid_street": str,
"resid_building_no": str,
"full_name": str,
"email": str,
"tos_accepted": bool
})
class RestMerchantEndpoints(Middleware):
@@ -55,13 +41,13 @@ class RestMerchantEndpoints(Middleware):
duration: Optional[int] = None,
webhook: Optional[str] = None,
redirect_url: Optional[str] = None) -> InvoiceSubmission:
body = _to_camel_case_keys({
"amount": amount, "currency": currency, "order_id": order_id,
"customer_info": customer_info, "pay_currencies": pay_currencies, "duration": duration,
"webhook": webhook, "redirect_url": redirect_url
})
body = {
"amount": amount, "currency": currency, "orderId": order_id,
"customerInfo": customer_info, "payCurrencies": pay_currencies, "duration": duration,
"webhook": webhook, "redirectUrl": redirect_url
}
data = _to_snake_case_keys(self._post("auth/w/ext/pay/invoice/create", body=body))
data = self._post("auth/w/ext/pay/invoice/create", body=body)
return InvoiceSubmission.parse(data)
@@ -76,9 +62,9 @@ class RestMerchantEndpoints(Middleware):
"limit": limit
}
response = self._post("auth/r/ext/pay/invoices", body=body)
data = self._post("auth/r/ext/pay/invoices", body=body)
return [ InvoiceSubmission.parse(sub_data) for sub_data in _to_snake_case_keys(response) ]
return [ InvoiceSubmission.parse(sub_data) for sub_data in data ]
def get_invoices_paginated(self,
page: int = 1,
@@ -91,13 +77,13 @@ class RestMerchantEndpoints(Middleware):
crypto: Optional[List[str]] = None,
id: Optional[str] = None,
order_id: Optional[str] = None) -> InvoicePage:
body = _to_camel_case_keys({
"page": page, "page_size": page_size, "sort": sort,
"sort_field": sort_field, "status": status, "fiat": fiat,
"crypto": crypto, "id": id, "order_id": order_id
})
body = {
"page": page, "pageSize": page_size, "sort": sort,
"sortField": sort_field, "status": status, "fiat": fiat,
"crypto": crypto, "id": id, "orderId": order_id
}
data = _to_snake_case_keys(self._post("auth/r/ext/pay/invoices/paginated", body=body))
data = self._post("auth/r/ext/pay/invoices/paginated", body=body)
return InvoicePage.parse(data)
@@ -105,13 +91,15 @@ class RestMerchantEndpoints(Middleware):
status: Literal["CREATED", "PENDING", "COMPLETED", "EXPIRED"],
format: str) -> List[InvoiceStats]:
return [ InvoiceStats(**sub_data) for sub_data in \
self._post("auth/r/ext/pay/invoice/stats/count", body={ "status": status, "format": format }) ]
self._post("auth/r/ext/pay/invoice/stats/count", \
body={ "status": status, "format": format }) ]
def get_invoice_earning_stats(self,
currency: str,
format: str) -> List[InvoiceStats]:
return [ InvoiceStats(**sub_data) for sub_data in \
self._post("auth/r/ext/pay/invoice/stats/earning", body={ "currency": currency, "format": format }) ]
self._post("auth/r/ext/pay/invoice/stats/earning", \
body={ "currency": currency, "format": format }) ]
def complete_invoice(self,
id: str,
@@ -119,45 +107,43 @@ class RestMerchantEndpoints(Middleware):
*,
deposit_id: Optional[int] = None,
ledger_id: Optional[int] = None) -> InvoiceSubmission:
return InvoiceSubmission.parse(_to_snake_case_keys(self._post("auth/w/ext/pay/invoice/complete", body={
body = {
"id": id, "payCcy": pay_currency, "depositId": deposit_id,
"ledgerId": ledger_id
})))
}
data = self._post("auth/w/ext/pay/invoice/complete", body=body)
return InvoiceSubmission.parse(data)
def expire_invoice(self, id: str) -> InvoiceSubmission:
body = { "id": id }
response = self._post("auth/w/ext/pay/invoice/expire", body=body)
return InvoiceSubmission.parse(_to_snake_case_keys(response))
data = self._post("auth/w/ext/pay/invoice/expire", body=body)
return InvoiceSubmission.parse(data)
def get_currency_conversion_list(self) -> List[CurrencyConversion]:
return [
CurrencyConversion(
base_currency=sub_data["baseCcy"],
convert_currency=sub_data["convertCcy"],
created=sub_data["created"]
) for sub_data in self._post("auth/r/ext/pay/settings/convert/list")
]
return [ CurrencyConversion(**sub_data) \
for sub_data in self._post("auth/r/ext/pay/settings/convert/list") ]
def add_currency_conversion(self,
base_currency: str,
convert_currency: str) -> bool:
return bool(self._post("auth/w/ext/pay/settings/convert/create", body={
"baseCcy": base_currency,
"convertCcy": convert_currency
}))
base_ccy: str,
convert_ccy: str) -> bool:
return bool(self._post("auth/w/ext/pay/settings/convert/create", \
body={ "baseCcy": base_ccy, "convertCcy": convert_ccy }))
def remove_currency_conversion(self,
base_currency: str,
convert_currency: str) -> bool:
return bool(self._post("auth/w/ext/pay/settings/convert/remove", body={
"baseCcy": base_currency,
"convertCcy": convert_currency
}))
base_ccy: str,
convert_ccy: str) -> bool:
return bool(self._post("auth/w/ext/pay/settings/convert/remove", \
body={ "baseCcy": base_ccy, "convertCcy": convert_ccy }))
def set_merchant_settings(self,
key: MerchantSettingsKey,
val: Any) -> bool:
return bool(self._post("auth/w/ext/pay/settings/set", body={ "key": key, "val": val }))
return bool(self._post("auth/w/ext/pay/settings/set", \
body={ "key": key, "val": val }))
def get_merchant_settings(self, key: MerchantSettingsKey) -> Any:
return self._post("auth/r/ext/pay/settings/get", body={ "key": key })
@@ -167,19 +153,28 @@ class RestMerchantEndpoints(Middleware):
def get_deposits(self,
start: int,
end: int,
to: int,
*,
ccy: Optional[str] = None,
unlinked: Optional[bool] = None) -> List[MerchantDeposit]:
body = { "from": start, "to": end, "ccy": ccy, "unlinked": unlinked }
response = self._post("auth/r/ext/pay/deposits", body=body)
return [ MerchantDeposit(**sub_data) for sub_data in _to_snake_case_keys(response) ]
body = {
"from": start, "to": to, "ccy": ccy,
"unlinked": unlinked
}
data = self._post("auth/r/ext/pay/deposits", body=body)
return [ MerchantDeposit(**sub_data) for sub_data in data ]
def get_unlinked_deposits(self,
ccy: str,
*,
start: Optional[int] = None,
end: Optional[int] = None) -> List[MerchantUnlinkedDeposit]:
body = { "ccy": ccy, "start": start, "end": end }
response = self._post("/auth/r/ext/pay/deposits/unlinked", body=body)
return [ MerchantUnlinkedDeposit(**sub_data) for sub_data in _to_snake_case_keys(response) ]
body = {
"ccy": ccy, "start": start, "end": end
}
data = self._post("/auth/r/ext/pay/deposits/unlinked", body=body)
return [ MerchantUnlinkedDeposit(**sub_data) for sub_data in data ]