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

Apply pylint's linting rules to bfxapi/rest/endpoints/rest_merchant_endpoints.py.

Browse Source
This commit is contained in:
Davide Casale
2023-03-07 18:12:32 +01:00
parent 06482ea3d3
commit 844c2526b9
1 changed files with 98 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,9 +1,15 @@
from typing import TypedDict, List, Union, Literal, Optional, Any #pylint: disable=redefined-builtin
from typing import TypedDict, Dict, List, Union, Literal, Optional, Any
from decimal import Decimal from decimal import Decimal
from .. types import * from .. types import \
InvoiceSubmission, InvoicePage, InvoiceStats, \
CurrencyConversion, MerchantDeposit, MerchantUnlinkedDeposit
from .. enums import MerchantSettingsKey from .. enums import MerchantSettingsKey
from .. middleware import Middleware from .. middleware import Middleware
from ...utils.camel_and_snake_case_helpers import to_snake_case_keys, to_camel_case_keys from ...utils.camel_and_snake_case_helpers import to_snake_case_keys, to_camel_case_keys
@@ -15,9 +21,17 @@ _CustomerInfo = TypedDict("_CustomerInfo", {
}) })
class RestMerchantEndpoints(Middleware): class RestMerchantEndpoints(Middleware):
def submit_invoice(self, amount: Union[Decimal, float, str], currency: str, order_id: str, #pylint: disable-next=too-many-arguments
customer_info: _CustomerInfo, pay_currencies: List[str], duration: Optional[int] = None, def submit_invoice(self,
webhook: Optional[str] = None, redirect_url: Optional[str] = None) -> InvoiceSubmission: amount: Union[Decimal, float, str],
currency: str,
order_id: str,
customer_info: _CustomerInfo,
pay_currencies: List[str],
*,
duration: Optional[int] = None,
webhook: Optional[str] = None,
redirect_url: Optional[str] = None) -> InvoiceSubmission:
body = to_camel_case_keys({ body = to_camel_case_keys({
"amount": amount, "currency": currency, "order_id": order_id, "amount": amount, "currency": currency, "order_id": order_id,
"customer_info": customer_info, "pay_currencies": pay_currencies, "duration": duration, "customer_info": customer_info, "pay_currencies": pay_currencies, "duration": duration,
@@ -25,18 +39,35 @@ class RestMerchantEndpoints(Middleware):
}) })
data = to_snake_case_keys(self._post("auth/w/ext/pay/invoice/create", body=body)) data = to_snake_case_keys(self._post("auth/w/ext/pay/invoice/create", body=body))
return InvoiceSubmission.parse(data) return InvoiceSubmission.parse(data)
def get_invoices(self, id: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[InvoiceSubmission]: def get_invoices(self,
return [ InvoiceSubmission.parse(sub_data) for sub_data in to_snake_case_keys(self._post("auth/r/ext/pay/invoices", body={ *,
id: Optional[str] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[InvoiceSubmission]:
body = {
"id": id, "start": start, "end": end, "id": id, "start": start, "end": end,
"limit": limit "limit": limit
})) ] }
def get_invoices_paginated(self, page: int = 1, page_size: int = 10, sort: Literal["asc", "desc"] = "asc", response = self._post("auth/r/ext/pay/invoices", body=body)
sort_field: Literal["t", "amount", "status"] = "t", status: Optional[List[Literal["CREATED", "PENDING", "COMPLETED", "EXPIRED"]]] = None, fiat: Optional[List[str]] = None,
crypto: Optional[List[str]] = None, id: Optional[str] = None, order_id: Optional[str] = None) -> InvoicePage: return [ InvoiceSubmission.parse(sub_data) for sub_data in to_snake_case_keys(response) ]
def get_invoices_paginated(self,
page: int = 1,
page_size: int = 10,
sort: Literal["asc", "desc"] = "asc",
sort_field: Literal["t", "amount", "status"] = "t",
*,
status: Optional[List[Literal["CREATED", "PENDING", "COMPLETED", "EXPIRED"]]] = None,
fiat: Optional[List[str]] = None,
crypto: Optional[List[str]] = None,
id: Optional[str] = None,
order_id: Optional[str] = None) -> InvoicePage:
body = to_camel_case_keys({ body = to_camel_case_keys({
"page": page, "page_size": page_size, "sort": sort, "page": page, "page_size": page_size, "sort": sort,
"sort_field": sort_field, "status": status, "fiat": fiat, "sort_field": sort_field, "status": status, "fiat": fiat,
@@ -44,60 +75,88 @@ class RestMerchantEndpoints(Middleware):
}) })
data = to_snake_case_keys(self._post("auth/r/ext/pay/invoices/paginated", body=body)) data = to_snake_case_keys(self._post("auth/r/ext/pay/invoices/paginated", body=body))
return InvoicePage.parse(data) return InvoicePage.parse(data)
def get_invoice_count_stats(self, status: Literal["CREATED", "PENDING", "COMPLETED", "EXPIRED"], format: str) -> List[InvoiceStats]: def get_invoice_count_stats(self,
return [ InvoiceStats(**sub_data) for sub_data in self._post("auth/r/ext/pay/invoice/stats/count", body={ "status": status, "format": format }) ] 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 }) ]
def get_invoice_earning_stats(self, currency: str, format: str) -> List[InvoiceStats]: def get_invoice_earning_stats(self,
return [ InvoiceStats(**sub_data) for sub_data in self._post("auth/r/ext/pay/invoice/stats/earning", body={ "currency": currency, "format": format }) ] 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 }) ]
def complete_invoice(self, id: str, pay_currency: str, deposit_id: Optional[int] = None, ledger_id: Optional[int] = None) -> InvoiceSubmission: def complete_invoice(self,
id: str,
pay_currency: str,
*,
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={ return InvoiceSubmission.parse(to_snake_case_keys(self._post("auth/w/ext/pay/invoice/complete", body={
"id": id, "payCcy": pay_currency, "depositId": deposit_id, "id": id, "payCcy": pay_currency, "depositId": deposit_id,
"ledgerId": ledger_id "ledgerId": ledger_id
}))) })))
def expire_invoice(self, id: str) -> InvoiceSubmission: def expire_invoice(self, id: str) -> InvoiceSubmission:
return InvoiceSubmission.parse(to_snake_case_keys(self._post("auth/w/ext/pay/invoice/expire", body={ "id": id }))) body = { "id": id }
response = self._post("auth/w/ext/pay/invoice/expire", body=body)
return InvoiceSubmission.parse(to_snake_case_keys(response))
def get_currency_conversion_list(self) -> List[CurrencyConversion]: def get_currency_conversion_list(self) -> List[CurrencyConversion]:
return [ return [
CurrencyConversion( CurrencyConversion(
base_currency=sub_data["baseCcy"], base_currency=sub_data["baseCcy"],
convert_currency=sub_data["convertCcy"], convert_currency=sub_data["convertCcy"],
created=sub_data["created"] created=sub_data["created"]
) for sub_data in self._post("auth/r/ext/pay/settings/convert/list") ) 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: def add_currency_conversion(self,
base_currency: str,
convert_currency: str) -> bool:
return bool(self._post("auth/w/ext/pay/settings/convert/create", body={ return bool(self._post("auth/w/ext/pay/settings/convert/create", body={
"baseCcy": base_currency, "baseCcy": base_currency,
"convertCcy": convert_currency "convertCcy": convert_currency
})) }))
def remove_currency_conversion(self, base_currency: str, convert_currency: str) -> bool: def remove_currency_conversion(self,
base_currency: str,
convert_currency: str) -> bool:
return bool(self._post("auth/w/ext/pay/settings/convert/remove", body={ return bool(self._post("auth/w/ext/pay/settings/convert/remove", body={
"baseCcy": base_currency, "baseCcy": base_currency,
"convertCcy": convert_currency "convertCcy": convert_currency
})) }))
def set_merchant_settings(self, key: MerchantSettingsKey, val: Any) -> bool: 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: def get_merchant_settings(self, key: MerchantSettingsKey) -> Any:
return self._post("auth/r/ext/pay/settings/get", body={ "key": key }) return self._post("auth/r/ext/pay/settings/get", body={ "key": key })
def list_merchant_settings(self, keys: List[MerchantSettingsKey] = list()) -> Dict[MerchantSettingsKey, Any]: def list_merchant_settings(self, keys: List[MerchantSettingsKey] = []) -> Dict[MerchantSettingsKey, Any]:
return self._post("auth/r/ext/pay/settings/list", body={ "keys": keys }) return self._post("auth/r/ext/pay/settings/list", body={ "keys": keys })
def get_deposits(self, start: int, end: int, ccy: Optional[str] = None, unlinked: Optional[bool] = None) -> List[MerchantDeposit]: def get_deposits(self,
return [ MerchantDeposit(**sub_data) for sub_data in to_snake_case_keys(self._post("auth/r/ext/pay/deposits", body={ start: int,
"from": start, "to": end, "ccy": ccy, "unlinked": unlinked end: int,
})) ] *,
ccy: Optional[str] = None,
def get_unlinked_deposits(self, ccy: str, start: Optional[int] = None, end: Optional[int] = None) -> List[MerchantUnlinkedDeposit]: unlinked: Optional[bool] = None) -> List[MerchantDeposit]:
return [ MerchantUnlinkedDeposit(**sub_data) for sub_data in to_snake_case_keys(self._post("/auth/r/ext/pay/deposits/unlinked", body={ body = { "from": start, "to": end, "ccy": ccy, "unlinked": unlinked }
"ccy": ccy, "start": start, "end": end response = self._post("auth/r/ext/pay/deposits", body=body)
})) ] return [ MerchantDeposit(**sub_data) for sub_data in to_snake_case_keys(response) ]
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) ]