From 8c07835f19edd3517258e2fe99f42d6ec0da3a0d Mon Sep 17 00:00:00 2001 From: cardosofede Date: Wed, 12 Jun 2024 11:35:09 +0200 Subject: [PATCH] (feat) abstract api calls from backend-api --- backend/services/backend_api_client.py | 116 ++++++++++++++----------- 1 file changed, 64 insertions(+), 52 deletions(-) diff --git a/backend/services/backend_api_client.py b/backend/services/backend_api_client.py index 935c387..df3ca61 100644 --- a/backend/services/backend_api_client.py +++ b/backend/services/backend_api_client.py @@ -1,3 +1,5 @@ +from typing import Optional, Dict + import pandas as pd import requests import streamlit as st @@ -13,7 +15,7 @@ class BackendAPIClient: _shared_instance = None @classmethod - def get_instance(cls, *args, **kwargs) -> "MarketsRecorder": + def get_instance(cls, *args, **kwargs) -> "BackendAPIClient": if cls._shared_instance is None: cls._shared_instance = BackendAPIClient(*args, **kwargs) return cls._shared_instance @@ -23,102 +25,112 @@ class BackendAPIClient: self.port = port self.base_url = f"http://{self.host}:{self.port}" + def post(self, endpoint: str, payload: Optional[Dict] = None, params: Optional[Dict] = None): + """ + Post request to the backend API. + :param params: + :param endpoint: + :param payload: + :return: + """ + url = f"{self.base_url}/{endpoint}" + response = requests.post(url, json=payload, params=params) + return self._process_response(response) + + def get(self, endpoint: str): + """ + Get request to the backend API. + :param endpoint: + :return: + """ + url = f"{self.base_url}/{endpoint}" + response = requests.get(url) + return self._process_response(response) + + @staticmethod + def _process_response(response): + if response.status_code == 400: + st.error(response.json()["detail"]) + return + return response.json() + def is_docker_running(self): """Check if Docker is running.""" - url = f"{self.base_url}/is-docker-running" - response = requests.get(url) - return response.json()["is_docker_running"] + endpoint = "is-docker-running" + return self.get(endpoint)["is_docker_running"] def pull_image(self, image_name: str): """Pull a Docker image.""" - url = f"{self.base_url}/pull-image/" - payload = {"image_name": image_name} - response = requests.post(url, json=payload) - return response.json() + endpoint = "pull-image" + return self.post(endpoint, payload={"image_name": image_name}) def list_available_images(self, image_name: str): """List available images by name.""" - url = f"{self.base_url}/available-images/{image_name}" - response = requests.get(url) - return response.json() + endpoint = f"available-images/{image_name}" + return self.get(endpoint) def list_active_containers(self): """List all active containers.""" - url = f"{self.base_url}/active-containers" - response = requests.get(url) - return response.json() + endpoint = "active-containers" + return self.get(endpoint) def list_exited_containers(self): """List all exited containers.""" - url = f"{self.base_url}/exited-containers" - response = requests.get(url) - return response.json() + endpoint = "exited-containers" + return self.get(endpoint) def clean_exited_containers(self): """Clean up exited containers.""" - url = f"{self.base_url}/clean-exited-containers" - response = requests.post(url) - return response.json() + endpoint = "clean-exited-containers" + return self.post(endpoint, payload=None) def remove_container(self, container_name: str, archive_locally: bool = True, s3_bucket: str = None): """Remove a specific container.""" - url = f"{self.base_url}/remove-container/{container_name}" + endpoint = f"remove-container/{container_name}" params = {"archive_locally": archive_locally} if s3_bucket: params["s3_bucket"] = s3_bucket - response = requests.post(url, params=params) - return response.json() + return self.post(endpoint, params=params) def stop_container(self, container_name: str): """Stop a specific container.""" - url = f"{self.base_url}/stop-container/{container_name}" - response = requests.post(url) - return response.json() + endpoint = f"stop-container/{container_name}" + return self.post(endpoint) def start_container(self, container_name: str): """Start a specific container.""" - url = f"{self.base_url}/start-container/{container_name}" - response = requests.post(url) - return response.json() + endpoint = f"start-container/{container_name}" + return self.post(endpoint) def create_hummingbot_instance(self, instance_config: dict): """Create a new Hummingbot instance.""" - url = f"{self.base_url}/create-hummingbot-instance" - response = requests.post(url, json=instance_config) - return response.json() + endpoint = "create-hummingbot-instance" + return self.post(endpoint, payload=instance_config) def start_bot(self, start_bot_config: dict): """Start a Hummingbot bot.""" - url = f"{self.base_url}/start-bot" - response = requests.post(url, json=start_bot_config) - return response.json() + endpoint = "start-bot" + return self.post(endpoint, payload=start_bot_config) def stop_bot(self, bot_name: str, skip_order_cancellation: bool = False, async_backend: bool = True): """Stop a Hummingbot bot.""" - url = f"{self.base_url}/stop-bot" - response = requests.post(url, json={"bot_name": bot_name, "skip_order_cancellation": skip_order_cancellation, "async_backend": async_backend}) - return response.json() + endpoint = "stop-bot" + return self.post(endpoint, payload={"bot_name": bot_name, "skip_order_cancellation": skip_order_cancellation, "async_backend": async_backend}) def import_strategy(self, strategy_config: dict): """Import a trading strategy to a bot.""" - url = f"{self.base_url}/import-strategy" - response = requests.post(url, json=strategy_config) - return response.json() + endpoint = "import-strategy" + return self.post(endpoint, payload=strategy_config) def get_bot_status(self, bot_name: str): """Get the status of a bot.""" - url = f"{self.base_url}/get-bot-status/{bot_name}" - response = requests.get(url) - if response.status_code == 200: - return response.json() - else: - return {"status": "error", "data": "Bot not found"} + endpoint = f"get-bot-status/{bot_name}" + self.get(endpoint) def get_bot_history(self, bot_name: str): """Get the historical data of a bot.""" - url = f"{self.base_url}/get-bot-history/{bot_name}" - response = requests.get(url) - return response.json() + endpoint = f"get-bot-history/{bot_name}" + return self.get(endpoint) def get_active_bots_status(self): """ @@ -286,8 +298,8 @@ class BackendAPIClient: response = requests.get(url) return response.json() - def get_all_balances(self): + def get_accounts_state(self): """Get all balances.""" - url = f"{self.base_url}/get-all-balances" + url = f"{self.base_url}/accounts-state" response = requests.get(url) return response.json()