khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
70a020f9b24f01a52603b0a4e66666ff60650177
api_utils_converse_v2/rest/controllers/async_base.py
T
khushalps 70a020f9b2 Squashed 'utils_v2/' content from commit 6c32fa9a 
git-subtree-dir: utils_v2
git-subtree-split: 6c32fa9a3d50c06a5487783423147b2bf9ce7549
2025-02-28 13:35:55 +05:30

461 lines
16 KiB
Python
Raw Blame History

"""
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.controllers.auth import RESTAuth
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 AsyncREST:
def __init__(
self,
http_client: httpx.AsyncClient = None,
debug = True,
debug_prefix = "REST (C) | ",
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,
auth: RESTAuth = None,
request_id: str | int = 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.
:param auth: Any authentication credentials that need to be sent.
:param request_id: An id to give this request. Can be useful for debugging later.
: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",
requestHeaders = headers,
requestParams = params,
requestId = request_id
)
try:
# Make the API call:
response = await self._http_client.get(
url = url,
headers = headers,
params = params,
auth = auth.encode()
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = response.reason_phrase
# Assume success if the response's HTTP code is in the 200 series:
api_response.success = response.is_success
# 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,
auth: RESTAuth = None,
request_id: str | int = 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).
:param auth: Any authentication credentials that need to be sent.
:param request_id: An id to give this request. Can be useful for debugging later.
: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",
requestHeaders = headers,
requestJson = json,
requestData = data,
requestContent = content,
requestId = request_id
)
try:
# Make the API call:
response = await self._http_client.post(
url = url,
headers = headers,
json = json,
data = data,
content = content,
auth = auth.encode()
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = response.reason_phrase
# Assume success if the response's HTTP code is in the 200 series:
api_response.success = response.is_success
# 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,
auth: RESTAuth = None,
request_id: str | int = 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.
:param auth: Any authentication credentials that need to be sent.
:param request_id: An id to give this request. Can be useful for debugging later.
: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",
requestHeaders = headers,
requestJson = json,
requestData = data,
requestId = request_id
)
try:
# Make the API call:
response = await self._http_client.put(
url = url,
headers = headers,
json = json,
data = data,
auth = auth.encode()
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = response.reason_phrase
# Assume success if the response's HTTP code is in the 200 series:
api_response.success = response.is_success
# 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 patch(
self,
url: str,
headers: dict = None,
json: dict = None,
data: dict = None,
content: str | bytes = None,
auth: RESTAuth = None,
request_id: str | int = None
) -> ApiResponse:
"""
To call an API using the PATCH 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).
:param auth: Any authentication credentials that need to be sent.
:param request_id: An id to give this request. Can be useful for debugging later.
: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 = "PATCH",
requestHeaders = headers,
requestJson = json,
requestData = data,
requestId = request_id
)
try:
# Make the API call:
response = await self._http_client.patch(
url = url,
headers = headers,
json = json,
data = data,
content = content,
auth = auth.encode()
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = response.reason_phrase
# Assume success if the response's HTTP code is in the 200 series:
api_response.success = response.is_success
# 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,
auth: RESTAuth = None,
request_id: str | int = None
) -> ApiResponse:
"""
To call an API using the DELETE method.
:param url: The URL to call.
:param headers: The headers to pass.
:param auth: Any authentication credentials that need to be sent.
:param request_id: An id to give this request. Can be useful for debugging later.
: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",
requestHeaders = headers,
requestId = request_id
)
try:
# Make the API call:
response = await self._http_client.delete(
url = url,
headers = headers,
auth = auth.encode()
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = response.reason_phrase
# Assume success if the response's HTTP code is in the 200 series:
api_response.success = response.is_success
# 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
Reference in New Issue View Git Blame Copy Permalink