From 06482ea3d3269fa9f850faf470956bdcf32de36f Mon Sep 17 00:00:00 2001 From: Davide Casale Date: Tue, 7 Mar 2023 18:02:26 +0100 Subject: [PATCH] Apply pylint's linting rules to bfxapi/rest/endpoints/rest_authenticated_endpoints.py. --- .../endpoints/rest_authenticated_endpoints.py | 407 ++++++++++++------ 1 file changed, 287 insertions(+), 120 deletions(-) diff --git a/bfxapi/rest/endpoints/rest_authenticated_endpoints.py b/bfxapi/rest/endpoints/rest_authenticated_endpoints.py index 0ed2241..97f85e7 100644 --- a/bfxapi/rest/endpoints/rest_authenticated_endpoints.py +++ b/bfxapi/rest/endpoints/rest_authenticated_endpoints.py @@ -1,42 +1,80 @@ -from typing import List, Tuple, Union, Literal, Optional +#pylint: disable=redefined-builtin + +from typing import Dict, List, Tuple, Union, Literal, Optional from decimal import Decimal from datetime import datetime -from .. types import * +from .. types import Notification, \ + UserInfo, LoginHistory, BalanceAvailable, \ + Order, Position, Trade, \ + FundingTrade, OrderTrade, Ledger, \ + FundingOffer, FundingCredit, FundingLoan, \ + FundingAutoRenew, FundingInfo, Wallet, \ + Transfer, Withdrawal, DepositAddress, \ + LightningNetworkInvoice, Movement, SymbolMarginInfo, \ + BaseMarginInfo, PositionClaim, PositionIncreaseInfo, \ + PositionIncrease, PositionHistory, PositionSnapshot, \ + PositionAudit, DerivativePositionCollateral, DerivativePositionCollateralLimits from .. import serializers + +from .. serializers import _Notification from .. enums import Sort, OrderType, FundingOfferType from .. middleware import Middleware +from ...utils.json_encoder import JSON + class RestAuthenticatedEndpoints(Middleware): def get_user_info(self) -> UserInfo: - return serializers.UserInfo.parse(*self._post(f"auth/r/info/user")) + return serializers.UserInfo.parse(*self._post("auth/r/info/user")) def get_login_history(self) -> List[LoginHistory]: - return [ serializers.LoginHistory.parse(*sub_data) for sub_data in self._post("auth/r/logins/hist") ] + return [ serializers.LoginHistory.parse(*sub_data) \ + for sub_data in self._post("auth/r/logins/hist") ] - def get_balance_available_for_orders_or_offers(self, symbol: str, type: str, dir: Optional[int] = None, rate: Optional[str] = None, lev: Optional[str] = None) -> BalanceAvailable: + def get_balance_available_for_orders_or_offers(self, + symbol: str, + type: str, + *, + dir: Optional[int] = None, + rate: Optional[str] = None, + lev: Optional[str] = None) -> BalanceAvailable: return serializers.BalanceAvailable.parse(*self._post("auth/calc/order/avail", body={ "symbol": symbol, "type": type, "dir": dir, "rate": rate, "lev": lev })) def get_wallets(self) -> List[Wallet]: - return [ serializers.Wallet.parse(*sub_data) for sub_data in self._post("auth/r/wallets") ] + return [ serializers.Wallet.parse(*sub_data) \ + for sub_data in self._post("auth/r/wallets") ] - def get_orders(self, symbol: Optional[str] = None, ids: Optional[List[str]] = None) -> List[Order]: + def get_orders(self, + *, + symbol: Optional[str] = None, + ids: Optional[List[str]] = None) -> List[Order]: endpoint = "auth/r/orders" - if symbol != None: + if symbol is not None: endpoint += f"/{symbol}" - return [ serializers.Order.parse(*sub_data) for sub_data in self._post(endpoint, body={ "id": ids }) ] + return [ serializers.Order.parse(*sub_data) \ + for sub_data in self._post(endpoint, body={ "id": ids }) ] - def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, float, str], - price: Optional[Union[Decimal, float, str]] = None, lev: Optional[int] = None, - price_trailing: Optional[Union[Decimal, float, str]] = None, price_aux_limit: Optional[Union[Decimal, float, str]] = None, price_oco_stop: Optional[Union[Decimal, float, str]] = None, - gid: Optional[int] = None, cid: Optional[int] = None, - flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None) -> Notification[Order]: + def submit_order(self, + type: OrderType, + symbol: str, + amount: Union[Decimal, float, str], + *, + price: Optional[Union[Decimal, float, str]] = None, + lev: Optional[int] = None, + price_trailing: Optional[Union[Decimal, float, str]] = None, + price_aux_limit: Optional[Union[Decimal, float, str]] = None, + price_oco_stop: Optional[Union[Decimal, float, str]] = None, + gid: Optional[int] = None, + cid: Optional[int] = None, + flags: Optional[int] = 0, + tif: Optional[Union[datetime, str]] = None, + meta: Optional[JSON] = None) -> Notification[Order]: body = { "type": type, "symbol": symbol, "amount": amount, "price": price, "lev": lev, @@ -44,32 +82,53 @@ class RestAuthenticatedEndpoints(Middleware): "gid": gid, "cid": cid, "flags": flags, "tif": tif, "meta": meta } - - return serializers._Notification[Order](serializers.Order).parse(*self._post("auth/w/order/submit", body=body)) - def update_order(self, id: int, amount: Optional[Union[Decimal, float, str]] = None, price: Optional[Union[Decimal, float, str]] = None, - cid: Optional[int] = None, cid_date: Optional[str] = None, gid: Optional[int] = None, - flags: Optional[int] = 0, lev: Optional[int] = None, delta: Optional[Union[Decimal, float, str]] = None, - price_aux_limit: Optional[Union[Decimal, float, str]] = None, price_trailing: Optional[Union[Decimal, float, str]] = None, tif: Optional[Union[datetime, str]] = None) -> Notification[Order]: + return _Notification[Order](serializers.Order).parse(*self._post("auth/w/order/submit", body=body)) + + def update_order(self, + id: int, + *, + amount: Optional[Union[Decimal, float, str]] = None, + price: Optional[Union[Decimal, float, str]] = None, + cid: Optional[int] = None, + cid_date: Optional[str] = None, + gid: Optional[int] = None, + flags: Optional[int] = 0, + lev: Optional[int] = None, + delta: Optional[Union[Decimal, float, str]] = None, + price_aux_limit: Optional[Union[Decimal, float, str]] = None, + price_trailing: Optional[Union[Decimal, float, str]] = None, + tif: Optional[Union[datetime, str]] = None) -> Notification[Order]: body = { "id": id, "amount": amount, "price": price, "cid": cid, "cid_date": cid_date, "gid": gid, "flags": flags, "lev": lev, "delta": delta, "price_aux_limit": price_aux_limit, "price_trailing": price_trailing, "tif": tif } - - return serializers._Notification[Order](serializers.Order).parse(*self._post("auth/w/order/update", body=body)) - def cancel_order(self, id: Optional[int] = None, cid: Optional[int] = None, cid_date: Optional[str] = None) -> Notification[Order]: - body = { + return _Notification[Order](serializers.Order) \ + .parse(*self._post("auth/w/order/update", body=body)) + + def cancel_order(self, + *, + id: Optional[int] = None, + cid: Optional[int] = None, + cid_date: Optional[str] = None) -> Notification[Order]: + body = { "id": id, "cid": cid, "cid_date": cid_date } - return serializers._Notification[Order](serializers.Order).parse(*self._post("auth/w/order/cancel", body=body)) + return _Notification[Order](serializers.Order) \ + .parse(*self._post("auth/w/order/cancel", body=body)) - def cancel_order_multi(self, ids: Optional[List[int]] = None, cids: Optional[List[Tuple[int, str]]] = None, gids: Optional[List[int]] = None, all: bool = False) -> Notification[List[Order]]: + def cancel_order_multi(self, + *, + ids: Optional[List[int]] = None, + cids: Optional[List[Tuple[int, str]]] = None, + gids: Optional[List[int]] = None, + all: bool = False) -> Notification[List[Order]]: body = { "ids": ids, "cids": cids, @@ -78,13 +137,20 @@ class RestAuthenticatedEndpoints(Middleware): "all": int(all) } - return serializers._Notification[List[Order]](serializers.Order, is_iterable=True).parse(*self._post("auth/w/order/cancel/multi", body=body)) + return _Notification[List[Order]](serializers.Order, is_iterable=True) \ + .parse(*self._post("auth/w/order/cancel/multi", body=body)) - def get_orders_history(self, symbol: Optional[str] = None, ids: Optional[List[int]] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Order]: - if symbol == None: + def get_orders_history(self, + *, + symbol: Optional[str] = None, + ids: Optional[List[int]] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[Order]: + if symbol is None: endpoint = "auth/r/orders/hist" else: endpoint = f"auth/r/orders/{symbol}/hist" - + body = { "id": ids, "start": start, "end": end, @@ -93,14 +159,23 @@ class RestAuthenticatedEndpoints(Middleware): return [ serializers.Order.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] - def get_order_trades(self, symbol: str, id: int) -> List[OrderTrade]: - return [ serializers.OrderTrade.parse(*sub_data) for sub_data in self._post(f"auth/r/order/{symbol}:{id}/trades") ] + def get_order_trades(self, + symbol: str, + id: int) -> List[OrderTrade]: + return [ serializers.OrderTrade.parse(*sub_data) \ + for sub_data in self._post(f"auth/r/order/{symbol}:{id}/trades") ] - def get_trades_history(self, symbol: Optional[str] = None, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Trade]: - if symbol == None: + def get_trades_history(self, + *, + symbol: Optional[str] = None, + sort: Optional[Sort] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[Trade]: + if symbol is None: endpoint = "auth/r/trades/hist" else: endpoint = f"auth/r/trades/{symbol}/hist" - + body = { "sort": sort, "start": start, "end": end, @@ -109,17 +184,24 @@ class RestAuthenticatedEndpoints(Middleware): return [ serializers.Trade.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] - def get_ledgers(self, currency: str, category: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Ledger]: + def get_ledgers(self, + currency: str, + *, + category: Optional[int] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[Ledger]: body = { "category": category, "start": start, "end": end, "limit": limit } - return [ serializers.Ledger.parse(*sub_data) for sub_data in self._post(f"auth/r/ledgers/{currency}/hist", body=body) ] + return [ serializers.Ledger.parse(*sub_data) \ + for sub_data in self._post(f"auth/r/ledgers/{currency}/hist", body=body) ] def get_base_margin_info(self) -> BaseMarginInfo: - return serializers.BaseMarginInfo.parse(*(self._post(f"auth/r/info/margin/base")[1])) + return serializers.BaseMarginInfo.parse(*(self._post("auth/r/info/margin/base")[1])) def get_symbol_margin_info(self, symbol: str) -> SymbolMarginInfo: response = self._post(f"auth/r/info/margin/{symbol}") @@ -127,51 +209,88 @@ class RestAuthenticatedEndpoints(Middleware): return serializers.SymbolMarginInfo.parse(*data) def get_all_symbols_margin_info(self) -> List[SymbolMarginInfo]: - return [ serializers.SymbolMarginInfo.parse(*([sub_data[1]] + sub_data[2])) for sub_data in self._post(f"auth/r/info/margin/sym_all") ] + return [ serializers.SymbolMarginInfo.parse(*([sub_data[1]] + sub_data[2])) \ + for sub_data in self._post("auth/r/info/margin/sym_all") ] def get_positions(self) -> List[Position]: - return [ serializers.Position.parse(*sub_data) for sub_data in self._post("auth/r/positions") ] + return [ serializers.Position.parse(*sub_data) for sub_data in self._post("auth/r/positions") ] - def claim_position(self, id: int, amount: Optional[Union[Decimal, float, str]] = None) -> Notification[PositionClaim]: - return serializers._Notification[PositionClaim](serializers.PositionClaim).parse( - *self._post("auth/w/position/claim", body={ "id": id, "amount": amount }) - ) + def claim_position(self, + id: int, + *, + amount: Optional[Union[Decimal, float, str]] = None) -> Notification[PositionClaim]: + return _Notification[PositionClaim](serializers.PositionClaim) \ + .parse(*self._post("auth/w/position/claim", body={ "id": id, "amount": amount })) - def increase_position(self, symbol: str, amount: Union[Decimal, float, str]) -> Notification[PositionIncrease]: - return serializers._Notification[PositionIncrease](serializers.PositionIncrease).parse( - *self._post("auth/w/position/increase", body={ "symbol": symbol, "amount": amount }) - ) + def increase_position(self, + symbol: str, + amount: Union[Decimal, float, str]) -> Notification[PositionIncrease]: + return _Notification[PositionIncrease](serializers.PositionIncrease) \ + .parse(*self._post("auth/w/position/increase", body={ "symbol": symbol, "amount": amount })) - def get_increase_position_info(self, symbol: str, amount: Union[Decimal, float, str]) -> PositionIncreaseInfo: - response = self._post(f"auth/r/position/increase/info", body={ "symbol": symbol, "amount": amount }) + def get_increase_position_info(self, + symbol: str, + amount: Union[Decimal, float, str]) -> PositionIncreaseInfo: + response = self._post("auth/r/position/increase/info", body={ "symbol": symbol, "amount": amount }) data = response[0] + [response[1][0]] + response[1][1] + [response[1][2]] + response[4] + response[5] return serializers.PositionIncreaseInfo.parse(*data) - def get_positions_history(self, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[PositionHistory]: - return [ serializers.PositionHistory.parse(*sub_data) for sub_data in self._post("auth/r/positions/hist", body={ "start": start, "end": end, "limit": limit }) ] + def get_positions_history(self, + *, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[PositionHistory]: + return [ serializers.PositionHistory.parse(*sub_data) \ + for sub_data in self._post("auth/r/positions/hist", \ + body={ "start": start, "end": end, "limit": limit }) ] - def get_positions_snapshot(self, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[PositionSnapshot]: - return [ serializers.PositionSnapshot.parse(*sub_data) for sub_data in self._post("auth/r/positions/snap", body={ "start": start, "end": end, "limit": limit }) ] + def get_positions_snapshot(self, + *, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[PositionSnapshot]: + return [ serializers.PositionSnapshot.parse(*sub_data) \ + for sub_data in self._post("auth/r/positions/snap", \ + body={ "start": start, "end": end, "limit": limit }) ] - def get_positions_audit(self, ids: Optional[List[int]] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[PositionAudit]: - return [ serializers.PositionAudit.parse(*sub_data) for sub_data in self._post("auth/r/positions/audit", body={ "ids": ids, "start": start, "end": end, "limit": limit }) ] + def get_positions_audit(self, + *, + ids: Optional[List[int]] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[PositionAudit]: + return [ serializers.PositionAudit.parse(*sub_data) \ + for sub_data in self._post("auth/r/positions/audit", \ + body={ "ids": ids, "start": start, "end": end, "limit": limit }) ] - def set_derivative_position_collateral(self, symbol: str, collateral: Union[Decimal, float, str]) -> DerivativePositionCollateral: - return serializers.DerivativePositionCollateral.parse(*(self._post("auth/w/deriv/collateral/set", body={ "symbol": symbol, "collateral": collateral })[0])) + def set_derivative_position_collateral(self, + symbol: str, + collateral: Union[Decimal, float, str]) -> DerivativePositionCollateral: + body = { "symbol": symbol, "collateral": collateral } + response = self._post("auth/w/deriv/collateral/set", body=body) + return serializers.DerivativePositionCollateral.parse(*(response)[0]) def get_derivative_position_collateral_limits(self, symbol: str) -> DerivativePositionCollateralLimits: - return serializers.DerivativePositionCollateralLimits.parse(*self._post("auth/calc/deriv/collateral/limits", body={ "symbol": symbol })) + body = { "symbol": symbol } + response = self._post("auth/calc/deriv/collateral/limit", body=body) + return serializers.DerivativePositionCollateralLimits.parse(*response) - def get_funding_offers(self, symbol: Optional[str] = None) -> List[FundingOffer]: + def get_funding_offers(self, *, symbol: Optional[str] = None) -> List[FundingOffer]: endpoint = "auth/r/funding/offers" - if symbol != None: + if symbol is not None: endpoint += f"/{symbol}" return [ serializers.FundingOffer.parse(*sub_data) for sub_data in self._post(endpoint) ] - def submit_funding_offer(self, type: FundingOfferType, symbol: str, amount: Union[Decimal, float, str], - rate: Union[Decimal, float, str], period: int, + #pylint: disable-next=too-many-arguments + def submit_funding_offer(self, + type: FundingOfferType, + symbol: str, + amount: Union[Decimal, float, str], + rate: Union[Decimal, float, str], + period: int, + *, flags: Optional[int] = 0) -> Notification[FundingOffer]: body = { "type": type, "symbol": symbol, "amount": amount, @@ -179,37 +298,52 @@ class RestAuthenticatedEndpoints(Middleware): "flags": flags } - return serializers._Notification[FundingOffer](serializers.FundingOffer).parse(*self._post("auth/w/funding/offer/submit", body=body)) + return _Notification[FundingOffer](serializers.FundingOffer) \ + .parse(*self._post("auth/w/funding/offer/submit", body=body)) def cancel_funding_offer(self, id: int) -> Notification[FundingOffer]: - return serializers._Notification[FundingOffer](serializers.FundingOffer).parse(*self._post("auth/w/funding/offer/cancel", body={ "id": id })) + return _Notification[FundingOffer](serializers.FundingOffer) \ + .parse(*self._post("auth/w/funding/offer/cancel", body={ "id": id })) def cancel_all_funding_offers(self, currency: str) -> Notification[Literal[None]]: - return serializers._Notification[Literal[None]](None).parse( - *self._post("auth/w/funding/offer/cancel/all", body={ "currency": currency }) - ) + return _Notification[Literal[None]](None) \ + .parse(*self._post("auth/w/funding/offer/cancel/all", body={ "currency": currency })) def submit_funding_close(self, id: int) -> Notification[Literal[None]]: - return serializers._Notification[Literal[None]](None).parse( - *self._post("auth/w/funding/close", body={ "id": id }) - ) + return _Notification[Literal[None]](None) \ + .parse(*self._post("auth/w/funding/close", body={ "id": id })) - def toggle_auto_renew(self, status: bool, currency: str, amount: Optional[str] = None, rate: Optional[int] = None, period: Optional[int] = None) -> Notification[FundingAutoRenew]: - return serializers._Notification[FundingAutoRenew](serializers.FundingAutoRenew).parse(*self._post("auth/w/funding/auto", body={ - "status": int(status), - "currency": currency, "amount": amount, - "rate": rate, "period": period - })) + def toggle_auto_renew(self, + status: bool, + currency: str, + *, + amount: Optional[str] = None, + rate: Optional[int] = None, + period: Optional[int] = None) -> Notification[FundingAutoRenew]: + return _Notification[FundingAutoRenew](serializers.FundingAutoRenew) \ + .parse(*self._post("auth/w/funding/auto", body={ + "status": int(status), "currency": currency, "amount": amount, + "rate": rate, "period": period + })) - def toggle_keep_funding(self, type: Literal["credit", "loan"], ids: Optional[List[int]] = None, changes: Optional[Dict[int, Literal[1, 2]]] = None) -> Notification[Literal[None]]: - return serializers._Notification[Literal[None]](None).parse(*self._post("auth/w/funding/keep", body={ + def toggle_keep_funding(self, + type: Literal["credit", "loan"], + *, + ids: Optional[List[int]] = None, + changes: Optional[Dict[int, Literal[1, 2]]] = None) -> Notification[Literal[None]]: + return _Notification[Literal[None]](None).parse(*self._post("auth/w/funding/keep", body={ "type": type, "id": ids, "changes": changes })) - def get_funding_offers_history(self, symbol: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingOffer]: - if symbol == None: + def get_funding_offers_history(self, + *, + symbol: Optional[str] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[FundingOffer]: + if symbol is None: endpoint = "auth/r/funding/offers/hist" else: endpoint = f"auth/r/funding/offers/{symbol}/hist" @@ -220,18 +354,23 @@ class RestAuthenticatedEndpoints(Middleware): return [ serializers.FundingOffer.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] - def get_funding_loans(self, symbol: Optional[str] = None) -> List[FundingLoan]: - if symbol == None: + def get_funding_loans(self, *, symbol: Optional[str] = None) -> List[FundingLoan]: + if symbol is None: endpoint = "auth/r/funding/loans" else: endpoint = f"auth/r/funding/loans/{symbol}" return [ serializers.FundingLoan.parse(*sub_data) for sub_data in self._post(endpoint) ] - def get_funding_loans_history(self, symbol: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingLoan]: - if symbol == None: + def get_funding_loans_history(self, + *, + symbol: Optional[str] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[FundingLoan]: + if symbol is None: endpoint = "auth/r/funding/loans/hist" else: endpoint = f"auth/r/funding/loans/{symbol}/hist" - + body = { "start": start, "end": end, "limit": limit @@ -239,18 +378,23 @@ class RestAuthenticatedEndpoints(Middleware): return [ serializers.FundingLoan.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] - def get_funding_credits(self, symbol: Optional[str] = None) -> List[FundingCredit]: - if symbol == None: + def get_funding_credits(self, *, symbol: Optional[str] = None) -> List[FundingCredit]: + if symbol is None: endpoint = "auth/r/funding/credits" else: endpoint = f"auth/r/funding/credits/{symbol}" return [ serializers.FundingCredit.parse(*sub_data) for sub_data in self._post(endpoint) ] - def get_funding_credits_history(self, symbol: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingCredit]: - if symbol == None: + def get_funding_credits_history(self, + *, + symbol: Optional[str] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[FundingCredit]: + if symbol is None: endpoint = "auth/r/funding/credits/hist" else: endpoint = f"auth/r/funding/credits/{symbol}/hist" - + body = { "start": start, "end": end, "limit": limit @@ -258,8 +402,14 @@ class RestAuthenticatedEndpoints(Middleware): return [ serializers.FundingCredit.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] - def get_funding_trades_history(self, symbol: Optional[str] = None, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingTrade]: - if symbol == None: + def get_funding_trades_history(self, + *, + symbol: Optional[str] = None, + sort: Optional[Sort] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[FundingTrade]: + if symbol is None: endpoint = "auth/r/funding/trades/hist" else: endpoint = f"auth/r/funding/trades/{symbol}/hist" @@ -276,46 +426,63 @@ class RestAuthenticatedEndpoints(Middleware): data = [response[1]] + response[2] return serializers.FundingInfo.parse(*data) - def transfer_between_wallets(self, from_wallet: str, to_wallet: str, currency: str, currency_to: str, amount: Union[Decimal, float, str]) -> Notification[Transfer]: + #pylint: disable-next=too-many-arguments + def transfer_between_wallets(self, + from_wallet: str, + to_wallet: str, + currency: str, + currency_to: str, + amount: Union[Decimal, float, str]) -> Notification[Transfer]: body = { "from": from_wallet, "to": to_wallet, "currency": currency, "currency_to": currency_to, "amount": amount } - return serializers._Notification[Transfer](serializers.Transfer).parse(*self._post("auth/w/transfer", body=body)) + return _Notification[Transfer](serializers.Transfer) \ + .parse(*self._post("auth/w/transfer", body=body)) - def submit_wallet_withdrawal(self, wallet: str, method: str, address: str, amount: Union[Decimal, float, str]) -> Notification[Withdrawal]: - return serializers._Notification[Withdrawal](serializers.Withdrawal).parse(*self._post("auth/w/withdraw", body={ - "wallet": wallet, "method": method, - "address": address, "amount": amount, + def submit_wallet_withdrawal(self, + wallet: str, + method: str, + address: str, + amount: Union[Decimal, float, str]) -> Notification[Withdrawal]: + return _Notification[Withdrawal](serializers.Withdrawal) \ + .parse(*self._post("auth/w/withdraw", body={ + "wallet": wallet, "method": method, "address": address, + "amount": amount, + })) + + def get_deposit_address(self, + wallet: str, + method: str, + renew: bool = False) -> Notification[DepositAddress]: + return _Notification[DepositAddress](serializers.DepositAddress) \ + .parse(*self._post("auth/w/deposit/address", body={ + "wallet": wallet, "method": method, "renew": int(renew) + })) + + def generate_deposit_invoice(self, + wallet: str, + currency: str, + amount: Union[Decimal, float, str]) -> LightningNetworkInvoice: + return serializers.LightningNetworkInvoice.parse(*self._post("auth/w/deposit/invoice", body={ + "wallet": wallet, "currency": currency, "amount": amount })) - def get_deposit_address(self, wallet: str, method: str, renew: bool = False) -> Notification[DepositAddress]: - body = { - "wallet": wallet, - "method": method, - "renew": int(renew) - } - - return serializers._Notification[DepositAddress](serializers.DepositAddress).parse(*self._post("auth/w/deposit/address", body=body)) - - def generate_deposit_invoice(self, wallet: str, currency: str, amount: Union[Decimal, float, str]) -> LightningNetworkInvoice: - body = { - "wallet": wallet, "currency": currency, - "amount": amount - } - - return serializers.LightningNetworkInvoice.parse(*self._post("auth/w/deposit/invoice", body=body)) - - def get_movements(self, currency: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Movement]: - if currency == None: + def get_movements(self, + *, + currency: Optional[str] = None, + start: Optional[str] = None, + end: Optional[str] = None, + limit: Optional[int] = None) -> List[Movement]: + if currency is None: endpoint = "auth/r/movements/hist" else: endpoint = f"auth/r/movements/{currency}/hist" - + body = { "start": start, "end": end, "limit": limit } - return [ serializers.Movement.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] \ No newline at end of file + return [ serializers.Movement.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ]