logos-messaging/logos-messaging-interop-tests
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-interop-tests.git synced 2026-01-10 01:43:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

revert base client

Browse Source
This commit is contained in:
Florin Barbu 2024-01-08 10:19:31 +02:00
parent 55fb702912
commit 6f062f7766
No known key found for this signature in database
GPG Key ID: 593D6DBC6D9E5095
2 changed files with 57 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
src/node/api_clients/base_client.py Normal file
View File

@ -0,0 +1,55 @@
import requests
from abc import ABC, abstractmethod
from src.env_vars import API_REQUEST_TIMEOUT
from src.libs.custom_logger import get_custom_logger
logger = get_custom_logger(__name__)
class BaseClient(ABC):
def make_request(self, method, url, headers=None, data=None):
logger.info(f"{method.upper()} call: {url} with payload: {data}")
response = requests.request(method.upper(), url, headers=headers, data=data, timeout=API_REQUEST_TIMEOUT)
try:
response.raise_for_status()
except requests.HTTPError as http_err:
logger.error(f"HTTP error occurred: {http_err}. Response content: {response.content}")
raise Exception(f"Error: {http_err} with response: {response.content}")
except Exception as err:
logger.error(f"An error occurred: {err}. Response content: {response.content}")
raise Exception(f"Error: {err} with response: {response.content}")
else:
logger.info(f"Response status code: {response.status_code}. Response content: {response.content}")
return response
@abstractmethod
def info(self):
pass
@abstractmethod
def set_relay_subscriptions(self, pubsub_topics):
pass
@abstractmethod
def delete_relay_subscriptions(self, pubsub_topics):
pass
@abstractmethod
def send_relay_message(self, message, pubsub_topic):
pass
@abstractmethod
def get_relay_messages(self, pubsub_topic):
pass
@abstractmethod
def set_filter_subscriptions(self, subscription):
pass
@abstractmethod
def delete_filter_subscriptions(self, subscription):
pass
@abstractmethod
def get_filter_messages(self, content_topic):
pass

20
src/node/api_clients/rest.py
View File

@ -1,31 +1,15 @@
import requests
from src.env_vars import API_REQUEST_TIMEOUT
from src.libs.custom_logger import get_custom_logger
import json
from urllib.parse import quote
from src.node.api_clients.base_client import BaseClient
logger = get_custom_logger(__name__)
class REST:
class REST(BaseClient):
def __init__(self, rest_port):
self._rest_port = rest_port
def make_request(self, method, url, headers=None, data=None):
logger.info(f"{method.upper()} call: {url} with payload: {data}")
response = requests.request(method.upper(), url, headers=headers, data=data, timeout=API_REQUEST_TIMEOUT)
try:
response.raise_for_status()
except requests.HTTPError as http_err:
logger.error(f"HTTP error occurred: {http_err}. Response content: {response.content}")
raise Exception(f"Error: {http_err} with response: {response.content}")
except Exception as err:
logger.error(f"An error occurred: {err}. Response content: {response.content}")
raise Exception(f"Error: {err} with response: {response.content}")
else:
logger.info(f"Response status code: {response.status_code}. Response content: {response.content}")
return response
def rest_call(self, method, endpoint, payload=None):
url = f"http://127.0.0.1:{self._rest_port}/{endpoint}"
headers = {"Content-Type": "application/json"}