khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Resetting utils subtree.

Browse Source
This commit is contained in:
khushal
2025-01-08 18:56:05 +05:30
parent 81fc28295f
commit 278951e8de
189 changed files with 80 additions and 142029 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/nse/controllers/base.py
-429
View File
@@ -1,429 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 28th Nov., 2024
OBJECTIVE:
To provide a base class for common behaviour of NSE's APIs.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
import asyncio
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
# My utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.nse.models.api_call import NSEApiResponse
# 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 AsyncNSEBase:
def __init__(
self,
base_url: str,
data_url: str,
http_client: httpx.AsyncClient = None,
cookies_refresh_interval: int | float = 300,
debug = True,
debug_prefix = "NSE | ",
debug_only_errors = True
):
"""
To be used as a base class for all sorts of NSE scraping.
:param base_url: The base URL (that you open in your browser) for your targeted module.
:param data_url: The internal URL (that NSE's script calls by itself) for your targeted module.
:param http_client: An HTTP client to use to make API calls.
:param cookies_refresh_interval: The no. of seconds after which the cookies should get refreshed.
:param debug: Whether, or not, you want to show debugging messages.
:param debug_prefix: The prefix string to use to recognize the module that is printing the debugging text.
:param debug_only_errors: Whether you want 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 the input configuration:
self._base_url = base_url
self._data_url = data_url
self._http_client = http_client or 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 = 2.5, # ... Time to wait for establishing a connection to the server.
write = 10.0, # .... Time to wait for sending data.
read = 60.0 # ...... Time to wait for receiving data.
)
)
# Make some provisions for later:
self._cookies = None
self._cookies_refresh_interval = cookies_refresh_interval
self._last_cookies_refresh = datetime.datetime.fromtimestamp(0, tz = date_time.TIMEZONE_UTC)
# For locking asynchronous tasks:
self._exclusive_lock = asyncio.Semaphore(1)
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
# ┓┏ ┓
# ┣┫┏┓┃┏┓┏┓┏┓┏
# ┛┗┗ ┗┣┛┗ ┛ ┛
# ┛
@staticmethod
def parse_datetime_string(
dt_str: str,
dt_formats: str | List[str] = None,
input_tz: str = date_time.TIMEZONE_IST,
output_tz: str = date_time.TIMEZONE_UTC
) -> datetime.datetime | None:
"""
Handles the date-time conversion and normalization for strings received from NSE.
:param dt_str: The string value as received from NSE.
:param dt_formats: The format(s) in which to attempt to parse the date string.
:param input_tz: The timezone in which the input must be assumed to be.
:param output_tz: The timezone the output must be converted to.
:return: The parsed datetime object, or null if parsing fails.
"""
# Basic input formatting:
dt_formats = dt_formats or "%d-%b-%Y"
# Try to parse the date-time string:
if isinstance(dt_formats, list):
dt_obj = date_time.parse_date_time(
input_value = dt_str,
date_formats = dt_formats
)
else:
try: dt_obj = datetime.datetime.strptime(dt_str, dt_formats)
except: dt_obj = None
# If the parsing failed:
if not dt_obj: return None
# If parsed successfully:
return date_time.to_timezone(
datetime_object = date_time.as_if_timezone(
datetime_object = dt_obj,
timezone = input_tz
),
timezone = output_tz
)
# ┏┓ ┓ • ┳┳┓ ┓ •
# ┃ ┏┓┏┓┃┏┓┏┓┏ ┃┃┃┏┓┃┏┓┏┓┏┓
# ┗┛┗┛┗┛┛┗┗┗ ┛ ┛ ┗┗┻┛┗┗┛┗┗┫
# ┛
@property
def seconds_since_cookies_refreshed(self):
return (date_time.get_current_utc_date_time() - self._last_cookies_refresh).total_seconds()
async def refresh_cookies(
self,
force_refresh = True,
retry_count: int = 1,
backoff_seconds: int | float = 0.5,
backoff_multiplier: float = 1.1
) -> bool:
"""
To get cookies for the given base URL. NSE has some sort of strict cookies and origin policy that is beyond my
current understanding. But I have noticed that if you furnish the right base URL and then use that, NSE will
respond just fine.
USAGE: Call this right after you initialize your class and then call it every so often after that.
:param force_refresh: Set to True to ignore the time criterion and refresh the cookie even if it is new.
:param retry_count: The no. of times to try to get the data from the API.
:param backoff_seconds: The delay between unsuccessful API calls.
:param backoff_multiplier: The multiplier to add to the delay to change delay.
:return: The cookies in the 'data' field of the
"""
if not self._debug_only_errors:
self._printer("Refreshing cookies.")
# Start by assuming failure:
success = False
# Ensure that only one cookie-refresh attempt is being made at a time:
async with self._exclusive_lock:
# We refresh the cookies only if the time has elapsed,
# or we have been asked to forcefully refresh the cookies:
if (
force_refresh or
self.seconds_since_cookies_refreshed > self._cookies_refresh_interval
):
# Make the API call:
api_response = await self.get(
url = self._base_url,
headers = {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0"
},
refresh_cookies = False,
force_refresh_cookies = False,
retry_count = retry_count,
backoff_seconds = backoff_seconds,
backoff_multiplier = backoff_multiplier
)
# If the API call succeeds:
if api_response.httpCode in [200]:
self._last_cookies_refresh = date_time.get_current_utc_date_time(as_string = False)
self._cookies = api_response.response.cookies
success = True
# Done here:
if not success: self._printer("Cookies NOT refreshed.")
elif not self._debug_only_errors: self._printer("Cookies refreshed.")
return success
# ┏┓┏┓┳ ┏┓ ┓┓•
# ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓
# ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫
# ┛
async def _get(
self,
url: str = None,
headers: dict = None,
params: dict = None,
cookies: httpx.Cookies | dict = None
) -> (httpx.Response | None, Exception | None):
"""
To call an API once using the GET method.
:param url: A custom URL to hit. If not specified, the data URL will be hit.
:param headers: Custom headers to use. If not specified, default values will be used.
:param params: The params to send in the query string itself.
:param cookies: The cookies to use when calling the API.
:return: The raw response from the API call and any exception that occurred.
"""
# Start by assuming failure:
response = None
excp = None
try:
# Make the API call:
response = await self._http_client.get(
url = url or self._data_url,
headers = headers or {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Connection": "keep-alive",
"Host": "www.nseindia.com",
"Referer": self._base_url,
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"Sec-GPC": "1",
"Upgrade-Insecure-Requests": "1",
"TE": "trailers",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0"
},
params = params,
cookies = cookies
)
# If something goes wrong:
except Exception as exception:
excp = exception
self._printer(exception, url, params)
# Done here:
return response, excp
async def get(
self,
url: str = None,
headers: dict = None,
params: dict = None,
cookies: httpx.Cookies | dict = None,
refresh_cookies: bool = True,
force_refresh_cookies: bool = False,
retry_count: int = 1,
backoff_seconds: int | float = 0.5,
backoff_multiplier: float = 1.1
) -> NSEApiResponse:
"""
Call an API using the GET method.
:param url: A custom URL to hit. If not specified, the data URL will be hit.
:param headers: Custom headers to use. If not specified, default values will be used.
:param params: The params to send in the query string itself.
:param cookies: Any custom cookies to use when calling the API. If not given, the class's internal cookies will
be used by default. Use this only to override the default one.
:param refresh_cookies: Whether, or not, you would like to refresh the cookies.
:param force_refresh_cookies: If set to True, cookies will be refreshed even if not timed out. If set to False,
cookies will be refreshed only when the interval passed to the constructor has elapsed since the last
successful refresh.
:param retry_count: The no. of times to try to get the data from the API.
:param backoff_seconds: The delay between unsuccessful API calls.
:param backoff_multiplier: The multiplier to add to the delay to change delay.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Select and refresh the cookies as needed:
if cookies: use_cookies = cookies
else:
if refresh_cookies: await self.refresh_cookies(force_refresh = force_refresh_cookies)
use_cookies = self._cookies
# Prepare the structure of the response:
api_response = NSEApiResponse(
action = inspect.stack()[1].function,
url = url or self._data_url,
method = "GET"
)
# Try as many times as asked:
for attempt in range(retry_count):
# Simple debugging statement:
if not self._debug_only_errors:
self._printer("API call.", attempt)
# Make the API call:
raw_response, exception = await self._get(
url = url,
headers = headers,
params = params,
cookies = use_cookies
)
# Populate the response model:
if raw_response is not None:
api_response.response = raw_response
api_response.httpCode = raw_response.status_code
api_response.message = raw_response.reason_phrase
if raw_response.status_code in [200]: api_response.success = True
if exception is not None:
api_response.exception = exception
api_response.message = str(exception)
# If the API call succeeded:
if api_response.success: break
# If the API call failed:
await asyncio.sleep(backoff_seconds)
backoff_seconds *= backoff_multiplier
# Done here:
return api_response
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass