61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Dict, Optional
|
|
import time
|
|
|
|
import requests
|
|
|
|
from cvttpy_tools.base.base import NamedObject
|
|
|
|
class RESTSender(NamedObject):
|
|
# Synchronous request sernder
|
|
session_: requests.Session
|
|
base_url_: str
|
|
|
|
def __init__(self, base_url: str) -> None:
|
|
self.base_url_ = base_url
|
|
self.session_ = requests.Session()
|
|
|
|
def is_ready(self) -> bool:
|
|
"""Checks if the server is up and responding"""
|
|
url = f"{self.base_url_}/ping"
|
|
try:
|
|
response = self.session_.get(url)
|
|
response.raise_for_status()
|
|
return True
|
|
except requests.exceptions.RequestException:
|
|
return False
|
|
|
|
def send_post(
|
|
self, endpoint: str, post_body: Dict, headers: Optional[Dict[str, str]] = None
|
|
) -> requests.Response:
|
|
|
|
if not headers:
|
|
headers = {"Content-Type": "application/json"}
|
|
url = f"{self.base_url_}/{endpoint}"
|
|
try:
|
|
return self.session_.request(
|
|
method="POST",
|
|
url=url,
|
|
json=post_body,
|
|
headers=headers,
|
|
)
|
|
except requests.exceptions.RequestException as excpt:
|
|
raise ConnectionError(
|
|
f"Failed to send status={excpt.response.status_code} {excpt.response.text}" # type: ignore
|
|
) from excpt
|
|
|
|
def send_get(
|
|
self, endpoint: str, headers: Optional[Dict[str, str]] = None
|
|
) -> requests.Response:
|
|
if not headers:
|
|
headers = {}
|
|
url = f"{self.base_url_}/{endpoint}"
|
|
try:
|
|
return self.session_.request(method="GET", url=url, headers=headers)
|
|
except requests.exceptions.RequestException as excpt:
|
|
raise ConnectionError(
|
|
f"Failed to send status={excpt.response.status_code} {excpt.response.text}" # type: ignore
|
|
) from excpt
|
|
|