You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1073 lines
35 KiB
Python
1073 lines
35 KiB
Python
"""See https://agoradesk.com/api-docs/v1."""
|
|
# pylint: disable=too-many-lines
|
|
# Large API. Lots of lines can't be avoided.
|
|
import json
|
|
import logging
|
|
from typing import Any, Dict, List, Optional, Union
|
|
|
|
import arrow
|
|
import treq
|
|
|
|
# Project imports
|
|
import util
|
|
from twisted.internet.defer import inlineCallbacks
|
|
|
|
__author__ = "marvin8"
|
|
__copyright__ = "(C) 2021 https://codeberg.org/MarvinsCryptoTools/agoradesk_py"
|
|
__version__ = "0.1.0"
|
|
|
|
# set logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
)
|
|
logging.getLogger("requests.packages.urllib3").setLevel(logging.INFO)
|
|
logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO)
|
|
logger = util.get_logger(__name__)
|
|
|
|
URI_API = "https://agoradesk.com/api/v1/"
|
|
|
|
|
|
class AgoraDesk:
|
|
"""AgoraDesk / LocalMonero API object.
|
|
|
|
Documentation: https://agoradesk.com/api-docs/v1
|
|
"""
|
|
|
|
# pylint: disable=too-many-public-methods
|
|
# API provides this many methods, I can't change that
|
|
|
|
def __init__(self, api_key: Optional[str], debug: Optional[bool] = False) -> None:
|
|
self.api_key = ""
|
|
if api_key:
|
|
self.api_key = api_key
|
|
self.debug = debug
|
|
|
|
if self.debug:
|
|
logging.getLogger("requests.packages.urllib3").setLevel(logging.DEBUG)
|
|
logger.setLevel(logging.DEBUG)
|
|
else:
|
|
logger.setLevel(logging.INFO)
|
|
|
|
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)
|
|
try:
|
|
text = yield response.content()
|
|
except: # noqa
|
|
self.log.error("Error with API call")
|
|
return
|
|
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,
|
|
http_method: Optional[str] = "GET",
|
|
query_values: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
api_call_url = URI_API + api_method
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"User-Agent": f"agoradesk_py/{__version__} "
|
|
f"https://codeberg.org/MarvinsCryptoTools/agoradesk_py",
|
|
"Authorization": self.api_key,
|
|
}
|
|
|
|
logger.debug("API Call URL: %s", api_call_url)
|
|
logger.debug("Headers : %s", headers)
|
|
logger.debug("HTTP Method : %s", http_method)
|
|
logger.debug("Query Values: %s", query_values)
|
|
logger.debug("Query Values as Json:\n%s", json.dumps(query_values))
|
|
|
|
result: Dict[str, Any] = {
|
|
"success": False,
|
|
"message": "Invalid Method",
|
|
"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).encode("ascii"),
|
|
)
|
|
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
|
|
|
|
# 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
|
|
# ===========================
|
|
|
|
def account_info(self, username: str) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getUserByUsername
|
|
"""
|
|
return self._api_call(api_method=f"account_info/{username}")
|
|
|
|
# def dashboard(self) -> Dict[str, Any]:
|
|
# """See Agoradesk API.
|
|
|
|
# https://agoradesk.com/api-docs/v1#operation/getUserDashboard
|
|
# """
|
|
# return self._api_call(api_method="dashboard")
|
|
|
|
def dashboard_buyer(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getUserDashboardBuyer
|
|
"""
|
|
return self._api_call(api_method="dashboard/buyer")
|
|
|
|
def dashboard(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getUserDashboardSeller
|
|
"""
|
|
return self._api_call(api_method="dashboard/seller")
|
|
|
|
def dashboard_canceled(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getUserDashboardCanceled
|
|
"""
|
|
return self._api_call(api_method="dashboard/canceled")
|
|
|
|
def dashboard_closed(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getUserDashboardClosed
|
|
"""
|
|
return self._api_call(api_method="dashboard/closed")
|
|
|
|
def dashboard_released(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getUserDashboardReleased
|
|
"""
|
|
|
|
return self._api_call(api_method="dashboard/released")
|
|
|
|
def logout(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/logout
|
|
"""
|
|
|
|
return self._api_call(api_method="logout", http_method="POST")
|
|
|
|
def myself(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getTokenOwnerUserData
|
|
"""
|
|
|
|
return self._api_call(api_method="myself")
|
|
|
|
def notifications(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getUserNotifications
|
|
"""
|
|
|
|
return self._api_call(api_method="notifications")
|
|
|
|
def notifications_mark_as_read(self, notification_id: str) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/markNotificationRead
|
|
"""
|
|
|
|
return self._api_call(
|
|
api_method=f"notifications/mark_as_read/{notification_id}",
|
|
http_method="POST",
|
|
)
|
|
|
|
def recent_messages(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getRecemtMessages
|
|
"""
|
|
|
|
return self._api_call(api_method="recent_messages")
|
|
|
|
# Trade related API Methods
|
|
# ===========================
|
|
|
|
# post/feedback/{username} • Give feedback to a user
|
|
def feedback(
|
|
self, username: str, feedback: str, msg: Optional[str]
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/setUserFeedback
|
|
"""
|
|
|
|
params = {"feedback": feedback}
|
|
if msg:
|
|
params["msg"] = msg
|
|
return self._api_call(
|
|
api_method=f"feedback/{username}",
|
|
http_method="POST",
|
|
query_values=params,
|
|
)
|
|
|
|
# Todo:
|
|
# post/trade/contact_release/{trade_id} • Release trade escrow
|
|
# post/contact_fund/{trade_id} • Fund a trade
|
|
# post/contact_dispute/{trade_id} • Start a trade dispute
|
|
|
|
# post/contact_mark_as_paid/{trade_id} • Mark a trade as paid
|
|
def contact_mark_as_paid(self, trade_id: str) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/markPaid
|
|
"""
|
|
return self._api_call(
|
|
api_method=f"contact_mark_as_paid/{trade_id}", http_method="POST"
|
|
)
|
|
|
|
# post/contact_cancel/{trade_id} • Cancel the trade
|
|
def contact_cancel(
|
|
self,
|
|
trade_id: str,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/cancelTrade
|
|
"""
|
|
return self._api_call(
|
|
api_method=f"contact_cancel/{trade_id}",
|
|
http_method="POST",
|
|
)
|
|
|
|
# Todo:
|
|
# post/contact_escrow/{trade_id} • Enable escrow
|
|
|
|
# get/contact_messages/{trade_id} • Get trade messages
|
|
def contact_messages(
|
|
self, trade_id: str, after: Optional[arrow.Arrow] = None
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getTradeMessages
|
|
"""
|
|
if after:
|
|
reply = self._api_call(
|
|
api_method=f"contact_messages/{trade_id}",
|
|
query_values={"after": after.to("UTC").isoformat()},
|
|
)
|
|
else:
|
|
reply = self._api_call(api_method=f"contact_messages/{trade_id}")
|
|
|
|
return reply
|
|
|
|
# post/contact_create/{ad_id} • Start a trade
|
|
def contact_create(
|
|
self,
|
|
ad_id: str,
|
|
amount: float,
|
|
msg: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/startTrade
|
|
"""
|
|
payload: Dict[str, Any] = {"amount": amount}
|
|
if msg:
|
|
payload["msg"] = msg
|
|
return self._api_call(
|
|
api_method=f"contact_create/{ad_id}",
|
|
http_method="POST",
|
|
query_values=payload,
|
|
)
|
|
|
|
# get/contact_info/{trade_id} • Get a trade by trade ID
|
|
def contact_info(self, trade_ids: Union[str, List[str]]) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getTradeById and
|
|
https://agoradesk.com/api-docs/v1#operation/getTradesInBulk
|
|
"""
|
|
api_method = "contact_info"
|
|
if isinstance(trade_ids, list):
|
|
params = "?contacts="
|
|
for trade_id in trade_ids:
|
|
params += f"{trade_id},"
|
|
params = params[0:-1]
|
|
else:
|
|
params = f"/{trade_ids}"
|
|
api_method += params
|
|
return self._api_call(api_method=api_method)
|
|
|
|
# Todo: Add image upload functionality
|
|
# post/contact_message_post/{trade_id} • Send a chat message/attachment
|
|
def contact_message_post(
|
|
self, trade_id: str, msg: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/sendChatMessage
|
|
"""
|
|
payload = {"msg": msg}
|
|
return self._api_call(
|
|
api_method=f"contact_message_post/{trade_id}",
|
|
http_method="POST",
|
|
query_values=payload,
|
|
)
|
|
|
|
# Todo:
|
|
# get/contact_message_attachment/{trade_id}/{attachment_id}
|
|
|
|
# Advertisement related API Methods
|
|
# ================================
|
|
|
|
def ad_create(
|
|
self,
|
|
country_code: str,
|
|
currency: str,
|
|
trade_type: str,
|
|
asset: str,
|
|
price_equation: str,
|
|
track_max_amount: bool,
|
|
require_trusted_by_advertiser: bool,
|
|
verified_email_required: Optional[bool] = None,
|
|
online_provider: Optional[str] = None,
|
|
msg: Optional[str] = None,
|
|
min_amount: Optional[float] = None,
|
|
max_amount: Optional[float] = None,
|
|
limit_to_fiat_amounts: Optional[str] = None,
|
|
payment_method_details: Optional[str] = None,
|
|
first_time_limit_asset: Optional[float] = None,
|
|
require_feedback_score: Optional[int] = None,
|
|
account_info: Optional[str] = None,
|
|
payment_window_minutes: Optional[int] = None,
|
|
floating: Optional[bool] = None,
|
|
lat: Optional[float] = None,
|
|
lon: Optional[float] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/createAd
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
# pylint: disable=too-many-locals
|
|
# pylint: disable=too-many-branches
|
|
# API takes this many arguments, I can't change that
|
|
# Too many locals and too many branches goes hand in hand
|
|
# with too many arguments
|
|
params: Dict[str, Any] = {
|
|
"countrycode": country_code,
|
|
"currency": currency,
|
|
"trade_type": trade_type,
|
|
"asset": asset,
|
|
"price_equation": price_equation,
|
|
"track_max_amount": 1 if track_max_amount else 0,
|
|
"require_trusted_by_advertiser": 1 if require_trusted_by_advertiser else 0,
|
|
}
|
|
if verified_email_required:
|
|
params["verified_email_required"] = 1 if verified_email_required else 0
|
|
if online_provider:
|
|
params["online_provider"] = online_provider
|
|
if msg:
|
|
params["msg"] = msg
|
|
if min_amount:
|
|
params["min_amount"] = min_amount
|
|
if max_amount:
|
|
params["max_amount"] = max_amount
|
|
if limit_to_fiat_amounts:
|
|
params["limit_to_fiat_amounts"] = limit_to_fiat_amounts
|
|
if payment_method_details:
|
|
params["payment_method_detail"] = payment_method_details
|
|
if first_time_limit_asset:
|
|
params["first_time_limit_asset"] = first_time_limit_asset
|
|
if require_feedback_score:
|
|
params["require_feedback_score"] = require_feedback_score
|
|
if account_info:
|
|
params["account_info"] = account_info
|
|
if payment_window_minutes:
|
|
params["payment_window_minutes"] = payment_window_minutes
|
|
if floating:
|
|
params["floating"] = 1 if floating else 0
|
|
if lat:
|
|
params["lat"] = lat
|
|
if lon:
|
|
params["lon"] = lon
|
|
|
|
return self._api_call(
|
|
api_method="ad-create",
|
|
http_method="POST",
|
|
query_values=params,
|
|
)
|
|
|
|
def ad(
|
|
self,
|
|
ad_id: str,
|
|
country_code: Optional[str] = None,
|
|
currency: Optional[str] = None,
|
|
trade_type: Optional[str] = None,
|
|
asset: Optional[str] = None,
|
|
price_equation: Optional[str] = None,
|
|
track_max_amount: Optional[bool] = None,
|
|
require_trusted_by_advertiser: Optional[bool] = None,
|
|
verified_email_required: Optional[bool] = None,
|
|
online_provider: Optional[str] = None,
|
|
msg: Optional[str] = None,
|
|
min_amount: Optional[float] = None,
|
|
max_amount: Optional[float] = None,
|
|
limit_to_fiat_amounts: Optional[str] = None,
|
|
payment_method_details: Optional[str] = None,
|
|
first_time_limit_asset: Optional[float] = None,
|
|
require_feedback_score: Optional[int] = None,
|
|
account_info: Optional[str] = None,
|
|
payment_window_minutes: Optional[int] = None,
|
|
floating: Optional[bool] = None,
|
|
lat: Optional[float] = None,
|
|
lon: Optional[float] = None,
|
|
visible: Optional[bool] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/editAd
|
|
"""
|
|
|
|
# pylint: disable=invalid-name
|
|
# Don't want to change the name of the method from what the API call is
|
|
# pylint: disable=too-many-arguments
|
|
# pylint: disable=too-many-locals
|
|
# pylint: disable=too-many-branches
|
|
# API takes this many arguments, I can't change that
|
|
# Too many locals and too many branches goes hand in hand
|
|
# with too many arguments
|
|
params: Dict[str, Union[str, float, bool]] = {}
|
|
if country_code:
|
|
params["countrycode"] = country_code
|
|
if currency:
|
|
params["currency"] = currency
|
|
if trade_type:
|
|
params["trade_type"] = trade_type
|
|
if asset:
|
|
params["asset"] = asset
|
|
if price_equation:
|
|
params["price_equation"] = price_equation
|
|
if track_max_amount:
|
|
params["track_max_amount"] = 1 if track_max_amount else 0
|
|
if require_trusted_by_advertiser:
|
|
params["require_trusted_by_advertiser"] = (
|
|
1 if require_trusted_by_advertiser else 0
|
|
)
|
|
if verified_email_required:
|
|
params["verified_email_required"] = 1 if verified_email_required else 0
|
|
if online_provider:
|
|
params["online_provider"] = online_provider
|
|
if msg:
|
|
params["msg"] = msg
|
|
if min_amount:
|
|
params["min_amount"] = min_amount
|
|
if max_amount:
|
|
params["max_amount"] = max_amount
|
|
if limit_to_fiat_amounts:
|
|
params["limit_to_fiat_amounts"] = limit_to_fiat_amounts
|
|
if payment_method_details:
|
|
params["payment_method_detail"] = payment_method_details
|
|
if first_time_limit_asset:
|
|
params["first_time_limit_asset"] = first_time_limit_asset
|
|
if require_feedback_score:
|
|
params["require_feedback_score"] = require_feedback_score
|
|
if account_info:
|
|
params["account_info"] = account_info
|
|
if payment_window_minutes:
|
|
params["payment_window_minutes"] = payment_window_minutes
|
|
if floating:
|
|
params["floating"] = 1 if floating else 0
|
|
if lat:
|
|
params["lat"] = lat
|
|
if lon:
|
|
params["lon"] = lon
|
|
if visible:
|
|
params["visible"] = True if visible else False
|
|
|
|
return self._api_call(
|
|
api_method=f"ad/{ad_id}",
|
|
http_method="POST",
|
|
query_values=params,
|
|
)
|
|
|
|
def ad_equation(self, ad_id: str, price_equation: str) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/updateFormula
|
|
"""
|
|
return self._api_call(
|
|
api_method=f"ad-equation/{ad_id}",
|
|
http_method="POST",
|
|
query_values={"price_equation": price_equation},
|
|
)
|
|
|
|
def ad_delete(self, ad_id: str) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/deleteAd
|
|
"""
|
|
return self._api_call(api_method=f"ad-delete/{ad_id}", http_method="POST")
|
|
|
|
def ads(
|
|
self,
|
|
country_code: Optional[str] = None,
|
|
currency: Optional[str] = None,
|
|
trade_type: Optional[str] = None,
|
|
visible: Optional[bool] = None,
|
|
asset: Optional[str] = None,
|
|
payment_method_code: Optional[str] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getYourAds
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
# API takes this many arguments, I can't change that
|
|
|
|
params = {}
|
|
if country_code:
|
|
params["countrycode"] = country_code
|
|
if currency:
|
|
params["currency"] = currency
|
|
if trade_type:
|
|
params["trade_type"] = trade_type
|
|
if visible is not None and visible:
|
|
params["visible"] = "1"
|
|
elif visible is not None and not visible:
|
|
params["visible"] = "0"
|
|
if asset:
|
|
params["asset"] = asset
|
|
if payment_method_code:
|
|
params["payment_method_code"] = payment_method_code
|
|
if page:
|
|
params["page"] = page
|
|
|
|
if len(params) == 0:
|
|
return self._api_call(api_method="ads")
|
|
|
|
return self._api_call(api_method="ads", query_values=params)
|
|
|
|
def ad_get(self, ad_ids: List[str]) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getAdById and
|
|
https://agoradesk.com/api-docs/v1#operation/getAdsInBulk
|
|
"""
|
|
api_method = "ad-get"
|
|
params = None
|
|
ids = str(ad_ids)[1:-1].replace(" ", "").replace("'", "")
|
|
|
|
if len(ad_ids) == 1:
|
|
api_method += f"/{ids}"
|
|
else:
|
|
params = {"ads": ids}
|
|
return self._api_call(api_method=api_method, query_values=params)
|
|
|
|
def payment_methods(self, country_code: Optional[str] = None) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/paymentMethods and
|
|
https://agoradesk.com/api-docs/v1#operation/countryHasPaymentMethod
|
|
"""
|
|
api_method = "payment_methods"
|
|
if country_code:
|
|
api_method += f"/{country_code}"
|
|
return self._api_call(api_method=api_method)
|
|
|
|
def country_codes(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/countryCodes
|
|
"""
|
|
return self._api_call(api_method="countrycodes")
|
|
|
|
def currencies(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/currencyCodes
|
|
"""
|
|
return self._api_call(api_method="currencies")
|
|
|
|
def equation(self, price_equation: str, currency: str) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/priceFormula
|
|
"""
|
|
return self._api_call(
|
|
api_method="equation",
|
|
http_method="POST",
|
|
query_values={
|
|
"price_equation": price_equation,
|
|
"currency": currency,
|
|
},
|
|
)
|
|
|
|
# Public ad search related API Methods
|
|
# ====================================
|
|
|
|
def _generic_online(
|
|
self,
|
|
direction: str,
|
|
main_currency: str,
|
|
exchange_currency: str,
|
|
country_code: Optional[str] = None,
|
|
payment_method: Optional[str] = None,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
# pylint: disable=too-many-arguments
|
|
|
|
add_to_api_method = ""
|
|
if country_code:
|
|
add_to_api_method = f"/{country_code}"
|
|
if payment_method:
|
|
add_to_api_method += f"/{payment_method}"
|
|
|
|
params = self._generic_search_parameters(amount, page)
|
|
return self._api_call(
|
|
api_method=f"{direction}-{main_currency}-online/"
|
|
f"{exchange_currency}{add_to_api_method}",
|
|
query_values=params,
|
|
)
|
|
|
|
@staticmethod
|
|
def _generic_search_parameters(amount, page):
|
|
params = None
|
|
if amount and page is not None:
|
|
params = {"amount": f"{amount}"}
|
|
elif amount and page is not None:
|
|
params = {"amount": f"{amount}", "page": f"{page}"}
|
|
elif not amount and page is not None:
|
|
params = {"page": f"{page}"}
|
|
return params
|
|
|
|
def buy_monero_online(
|
|
self,
|
|
currency_code: str,
|
|
country_code: Optional[str] = None,
|
|
payment_method: Optional[str] = None,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineSellXmrByCurrencyCode and
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineSellXmrByCurrencyCodeAndCountryCode and # noqa: E501 pylint: disable=line-too-long
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineSellXmrByCurrencyCodeAndPaymentMethodCode and # noqa: E501 pylint: disable=line-too-long
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineSellXmrByCurrencyCodeAndCountryCodeAndPaymentMethodCode # noqa: E501 pylint: disable=line-too-long
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
|
|
return self._generic_online(
|
|
direction="buy",
|
|
main_currency="monero",
|
|
exchange_currency=currency_code,
|
|
country_code=country_code,
|
|
payment_method=payment_method,
|
|
amount=amount,
|
|
page=page,
|
|
)
|
|
|
|
def buy_bitcoins_online(
|
|
self,
|
|
currency_code: str,
|
|
country_code: Optional[str] = None,
|
|
payment_method: Optional[str] = None,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineSellBtcByCurrencyCode
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineSellBtcByCurrencyCodeAndCountryCode and # noqa: E501 pylint: disable=line-too-long
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineSellBtcByCurrencyCodeAndPaymentMethodCode and # noqa: E501 pylint: disable=line-too-long
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineSellBtcByCurrencyCodeAndCountryCodeAndPaymentMethodCode # noqa: E501 pylint: disable=line-too-long
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
|
|
return self._generic_online(
|
|
direction="buy",
|
|
main_currency="bitcoins",
|
|
exchange_currency=currency_code,
|
|
country_code=country_code,
|
|
payment_method=payment_method,
|
|
amount=amount,
|
|
page=page,
|
|
)
|
|
|
|
def sell_monero_online(
|
|
self,
|
|
currency_code: str,
|
|
country_code: Optional[str] = None,
|
|
payment_method: Optional[str] = None,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineBuyXmrByCurrencyCode
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineBuyXmrByCurrencyCodeAndCountryCode and # noqa: E501 pylint: disable=line-too-long
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineBuyXmrByCurrencyCodeAndPaymentMethodCode and # noqa: E501 pylint: disable=line-too-long
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineBuyXmrByCurrencyCodeAndCountryCodeAndPaymentMethodCode # noqa: E501 pylint: disable=line-too-long
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
|
|
return self._generic_online(
|
|
direction="sell",
|
|
main_currency="monero",
|
|
exchange_currency=currency_code,
|
|
country_code=country_code,
|
|
payment_method=payment_method,
|
|
amount=amount,
|
|
page=page,
|
|
)
|
|
|
|
def sell_bitcoins_online(
|
|
self,
|
|
currency_code: str,
|
|
country_code: Optional[str] = None,
|
|
payment_method: Optional[str] = None,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineBuyBtcByCurrencyCode
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineBuyBtcByCurrencyCodeAndCountryCode and # noqa: E501 pylint: disable=line-too-long
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineBuyBtcByCurrencyCodeAndPaymentMethodCode and # noqa: E501 pylint: disable=line-too-long
|
|
https://agoradesk.com/api-docs/v1#operation/getOnlineBuyBtcByCurrencyCodeAndCountryCodeAndPaymentMethodCode # noqa: E501 pylint: disable=line-too-long
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
|
|
return self._generic_online(
|
|
direction="sell",
|
|
main_currency="bitcoins",
|
|
exchange_currency=currency_code,
|
|
country_code=country_code,
|
|
payment_method=payment_method,
|
|
amount=amount,
|
|
page=page,
|
|
)
|
|
|
|
def _generic_cash(
|
|
self,
|
|
direction: str,
|
|
main_currency: str,
|
|
exchange_currency: str,
|
|
country_code: str,
|
|
lat: str,
|
|
lon: str,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
# pylint: disable=too-many-arguments
|
|
|
|
params = self._generic_search_parameters(amount, page)
|
|
|
|
return self._api_call(
|
|
api_method=f"{direction}-{main_currency}-with-cash/"
|
|
f"{exchange_currency}/{country_code}/{lat}/{lon}",
|
|
query_values=params,
|
|
)
|
|
|
|
def buy_monero_with_cash(
|
|
self,
|
|
currency_code: str,
|
|
country_code: str,
|
|
lat: str,
|
|
lon: str,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getLocalSellXmrByCurrencyCodeAndCountryCode # noqa: E501 pylint: disable=line-too-long
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
|
|
return self._generic_cash(
|
|
direction="buy",
|
|
main_currency="monero",
|
|
exchange_currency=currency_code,
|
|
country_code=country_code,
|
|
lat=lat,
|
|
lon=lon,
|
|
amount=amount,
|
|
page=page,
|
|
)
|
|
|
|
def buy_bitcoins_with_cash(
|
|
self,
|
|
currency_code: str,
|
|
country_code: str,
|
|
lat: str,
|
|
lon: str,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getLocalSellBtcByCurrencyCodeAndCountryCode # noqa: E501 pylint: disable=line-too-long
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
|
|
return self._generic_cash(
|
|
direction="buy",
|
|
main_currency="bitcoins",
|
|
exchange_currency=currency_code,
|
|
country_code=country_code,
|
|
lat=lat,
|
|
lon=lon,
|
|
amount=amount,
|
|
page=page,
|
|
)
|
|
|
|
def sell_monero_with_cash(
|
|
self,
|
|
currency_code: str,
|
|
country_code: str,
|
|
lat: str,
|
|
lon: str,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getLocalBuyXmrByCurrencyCodeAndCountryCode # noqa: E501 pylint: disable=line-too-long
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
|
|
return self._generic_cash(
|
|
direction="sell",
|
|
main_currency="monero",
|
|
exchange_currency=currency_code,
|
|
country_code=country_code,
|
|
lat=lat,
|
|
lon=lon,
|
|
amount=amount,
|
|
page=page,
|
|
)
|
|
|
|
def sell_bitcoins_with_cash(
|
|
self,
|
|
currency_code: str,
|
|
country_code: str,
|
|
lat: str,
|
|
lon: str,
|
|
amount: Optional[float] = None,
|
|
page: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getLocalBuyBtcByCurrencyCodeAndCountryCode # noqa: E501 pylint: disable=line-too-long
|
|
"""
|
|
|
|
# pylint: disable=too-many-arguments
|
|
|
|
return self._generic_cash(
|
|
direction="sell",
|
|
main_currency="bitcoins",
|
|
exchange_currency=currency_code,
|
|
country_code=country_code,
|
|
lat=lat,
|
|
lon=lon,
|
|
amount=amount,
|
|
page=page,
|
|
)
|
|
|
|
# Statistics related API Methods
|
|
# ==============================
|
|
|
|
def moneroaverage(
|
|
self, currency: Optional[str] = "ticker-all-currencies"
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getXmrTicker and
|
|
https://agoradesk.com/api-docs/v1#operation/getXmrTickerByCurrencyCode
|
|
"""
|
|
return self._api_call(api_method=f"moneroaverage/{currency}")
|
|
|
|
# Wallet related API Methods
|
|
# ===========================
|
|
|
|
def wallet(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getBtcWallet
|
|
"""
|
|
return self._api_call(api_method="wallet")
|
|
|
|
def wallet_balance(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getBtcWalletBalance
|
|
"""
|
|
return self._api_call(api_method="wallet-balance")
|
|
|
|
def wallet_xmr(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getXmrWallet
|
|
"""
|
|
return self._api_call(api_method="wallet/XMR")
|
|
|
|
def wallet_balance_xmr(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getXmrWalletBalance
|
|
"""
|
|
return self._api_call(api_method="wallet-balance/XMR")
|
|
|
|
def wallet_addr(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getBtcAddress
|
|
"""
|
|
return self._api_call(api_method="wallet-addr")
|
|
|
|
def wallet_addr_xmr(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getXMRAddress
|
|
"""
|
|
return self._api_call(api_method="wallet-addr/XMR")
|
|
|
|
def fees(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getBtcFee
|
|
"""
|
|
return self._api_call(api_method="fees")
|
|
|
|
def fees_xmr(self) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/getXmrFee
|
|
"""
|
|
return self._api_call(api_method="fees/XMR")
|
|
|
|
def wallet_send(
|
|
self,
|
|
address: str,
|
|
amount: float,
|
|
password: str,
|
|
fee_level: str,
|
|
otp: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/withdrawBtc
|
|
"""
|
|
# pylint: disable=too-many-arguments
|
|
|
|
params = {
|
|
"address": address,
|
|
"amount": amount,
|
|
"password": password,
|
|
"fee_level": fee_level,
|
|
}
|
|
if otp:
|
|
params["otp"] = otp
|
|
|
|
return self._api_call(
|
|
api_method="wallet-send", http_method="POST", query_values=params
|
|
)
|
|
|
|
def wallet_send_xmr(
|
|
self,
|
|
address: str,
|
|
amount: float,
|
|
password: str,
|
|
# fee_level: str,
|
|
otp: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""See Agoradesk API.
|
|
|
|
https://agoradesk.com/api-docs/v1#operation/withdrawXmr
|
|
"""
|
|
# pylint: disable=too-many-arguments
|
|
|
|
params = {
|
|
"address": address,
|
|
"amount": amount,
|
|
"password": password,
|
|
# "fee_level": fee_level,
|
|
}
|
|
if otp:
|
|
params["otp"] = otp
|
|
|
|
return self._api_call(
|
|
api_method="wallet-send/XMR",
|
|
http_method="POST",
|
|
query_values=params,
|
|
)
|