khushalps
d5dc737b86
git-subtree-dir: utils_v2 git-subtree-split: 97ea8e2eee0b34c9bf906d60cd7fe14b4ef82daa
351 lines
12 KiB
Python
351 lines
12 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Monday, 16th Dec., 2024
|
|
|
|
OBJECTIVE:
|
|
|
|
To provide a class to make REST-ful API calls and have a structured approach for the inputs and outputs.
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# System-level activities:
|
|
import io
|
|
|
|
# The base model:
|
|
from utils_v2.rest.models.api_call import ApiResponse
|
|
|
|
# My utils:
|
|
from utils_v2.date_time import date_time
|
|
|
|
# To make API calls:
|
|
import httpx
|
|
|
|
# To work with date and time:
|
|
import datetime
|
|
|
|
# For working with datatypes:
|
|
from typing import Literal, List
|
|
|
|
# For debugging:
|
|
from icecream import IceCreamDebugger
|
|
import inspect
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
class AsyncRestBase:
|
|
|
|
def __init__(
|
|
self,
|
|
http_client: httpx.AsyncClient = None,
|
|
debug = True,
|
|
debug_prefix = "GMail | ",
|
|
debug_only_errors = True
|
|
):
|
|
|
|
"""
|
|
To initialize the base class.
|
|
:param http_client: An asynchronous HTTP client to make API calls.
|
|
:param debug: Whether, or not, you would like to show debugging messages on the terminal.
|
|
:param debug_prefix: The prefix string to identify the debugging messages.
|
|
:param debug_only_errors: Whether you would like to show all debugging messages or just error messages.
|
|
"""
|
|
|
|
# Prepare the debugging utility:
|
|
self._debug_prefix = debug_prefix
|
|
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
|
|
if not debug: self._printer.disable()
|
|
self._debug_only_errors = debug_only_errors
|
|
|
|
# Accept/create an HTTP client to work with:
|
|
if http_client: self._http_client = http_client
|
|
else: self._http_client = httpx.AsyncClient(
|
|
limits = httpx.Limits(
|
|
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
|
|
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
|
|
),
|
|
timeout = httpx.Timeout(
|
|
pool = 120.0, # .... Time to wait for a free connection from the pool.
|
|
connect = 5.0, # ... Time to wait for establishing a connection to the server.
|
|
write = 10.0, # .... Time to wait for sending data.
|
|
read = 120.0 # ..... Time to wait for receiving data.
|
|
)
|
|
)
|
|
|
|
def enable_debug(self):
|
|
self._printer.enable()
|
|
|
|
def disable_debug(self):
|
|
self._printer.disable()
|
|
|
|
def debug_only_errors(self):
|
|
self._debug_only_errors = True
|
|
|
|
def debug_everything(self):
|
|
self._debug_only_errors = False
|
|
|
|
# ┏┓┏┓┳ ┏┓ ┓┓•
|
|
# ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓
|
|
# ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫
|
|
# ┛
|
|
|
|
async def get(
|
|
self,
|
|
url: str,
|
|
headers: dict = None,
|
|
params: dict = None
|
|
) -> ApiResponse:
|
|
|
|
"""
|
|
To call an API using the GET method.
|
|
:param url: The URL to call.
|
|
:param headers: The headers to pass.
|
|
:param params: The params to send in the query string itself.
|
|
:return: A structured response that includes the raw response, the exception (if any), and so on.
|
|
"""
|
|
|
|
# Prepare the structure of the response:
|
|
api_response = ApiResponse(
|
|
action = inspect.stack()[1].function,
|
|
url = url,
|
|
method = "GET"
|
|
)
|
|
|
|
try:
|
|
|
|
# Make the API call:
|
|
response = await self._http_client.get(
|
|
url = url,
|
|
headers = headers,
|
|
params = params
|
|
)
|
|
|
|
# Note down the results:
|
|
api_response.response = response
|
|
api_response.httpCode = response.status_code
|
|
api_response.message = response.reason_phrase
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
api_response.exception = exception
|
|
api_response.message = str(exception)
|
|
self._printer(exception, api_response.url, api_response.method, headers, params)
|
|
|
|
# Done here:
|
|
return api_response
|
|
|
|
async def post(
|
|
self,
|
|
url: str,
|
|
headers: dict = None,
|
|
json: dict = None,
|
|
data: dict = None,
|
|
content: str | bytes = None
|
|
) -> ApiResponse:
|
|
|
|
"""
|
|
To call an API using the POST method.
|
|
:param url: The URL to call.
|
|
:param headers: The headers to pass.
|
|
:param json: The params to send in the JSON body.
|
|
:param data: The params to send in the form-data in the body.
|
|
:param content: The raw content to be sent in the body (typically as an octet-stream).
|
|
:return: A structured response that includes the raw response, the exception (if any), and so on.
|
|
"""
|
|
|
|
# Prepare the structure of the response:
|
|
api_response = ApiResponse(
|
|
action = inspect.stack()[1].function,
|
|
url = url,
|
|
method = "POST"
|
|
)
|
|
|
|
try:
|
|
|
|
# Make the API call:
|
|
response = await self._http_client.post(
|
|
url = url,
|
|
headers = headers,
|
|
json = json,
|
|
data = data,
|
|
content = content
|
|
)
|
|
|
|
# Note down the results:
|
|
api_response.response = response
|
|
api_response.httpCode = response.status_code
|
|
api_response.message = response.reason_phrase
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
api_response.exception = exception
|
|
api_response.message = str(exception)
|
|
self._printer(exception, api_response.url, api_response.method, headers, json, data)
|
|
|
|
# Done here:
|
|
return api_response
|
|
|
|
async def put(
|
|
self,
|
|
url: str,
|
|
headers: dict = None,
|
|
json: dict = None,
|
|
data: dict = None
|
|
) -> ApiResponse:
|
|
|
|
"""
|
|
To call an API using the PUT method.
|
|
:param url: The URL to call.
|
|
:param headers: The headers to pass.
|
|
:param json: The params to send in the JSON body.
|
|
:param data: The params to send in the form-data in the body.
|
|
:return: A structured response that includes the raw response, the exception (if any), and so on.
|
|
"""
|
|
|
|
# Prepare the structure of the response:
|
|
api_response = ApiResponse(
|
|
action = inspect.stack()[1].function,
|
|
url = url,
|
|
method = "PUT"
|
|
)
|
|
|
|
try:
|
|
|
|
# Make the API call:
|
|
response = await self._http_client.put(
|
|
url = url,
|
|
headers = headers,
|
|
json = json,
|
|
data = data
|
|
)
|
|
|
|
# Note down the results:
|
|
api_response.response = response
|
|
api_response.httpCode = response.status_code
|
|
api_response.message = response.reason_phrase
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
api_response.exception = exception
|
|
api_response.message = str(exception)
|
|
self._printer(exception, api_response.url, api_response.method, headers, json, data)
|
|
|
|
# Done here:
|
|
return api_response
|
|
|
|
async def delete(
|
|
self,
|
|
url: str,
|
|
headers: dict = None
|
|
) -> ApiResponse:
|
|
|
|
"""
|
|
To call an API using the DELETE method.
|
|
:param url: The URL to call.
|
|
:param headers: The headers to pass.
|
|
:return: A structured response that includes the raw response, the exception (if any), and so on.
|
|
"""
|
|
|
|
# Prepare the structure of the response:
|
|
api_response = ApiResponse(
|
|
action = inspect.stack()[1].function,
|
|
url = url,
|
|
method = "DELETE"
|
|
)
|
|
|
|
try:
|
|
|
|
# Make the API call:
|
|
response = await self._http_client.delete(
|
|
url = url,
|
|
headers = headers
|
|
)
|
|
|
|
# Note down the results:
|
|
api_response.response = response
|
|
api_response.httpCode = response.status_code
|
|
api_response.message = response.reason_phrase
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
api_response.exception = exception
|
|
api_response.message = str(exception)
|
|
self._printer(exception, api_response.url, api_response.method, headers)
|
|
|
|
# Done here:
|
|
return api_response
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
pass
|