""" 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 = 5.0, # ... 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 # ┏┓ • # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏ # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛ # ┛ @property def cookies(self): return self._cookies # ┓┏ ┓ # ┣┫┏┓┃┏┓┏┓┏┓┏ # ┛┗┗ ┗┣┛┗ ┛ ┛ # ┛ @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 """ # Start by assuming failure: success = False http_code = None # 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._cookies is None 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", "Upgrade-Insecure-Requests": "1", "Sec-Ch-Ua": "\"Chromium\";v=\"128\", \"Not;A=Brand\";v=\"24\", \"Google Chrome\";v=\"128\"", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36" }, refresh_cookies = False, force_refresh_cookies = False, retry_count = retry_count, backoff_seconds = backoff_seconds, backoff_multiplier = backoff_multiplier ) http_code = api_response.httpCode # 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 or not self._debug_only_errors: self._printer(success, http_code) 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", "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", "Sec-Ch-Ua": "\"Chromium\";v=\"128\", \"Not;A=Brand\";v=\"24\", \"Google Chrome\";v=\"128\"", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36" }, 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: frame = inspect.stack()[1] caller_locals = frame[0].f_locals calling_class = caller_locals.get("self", None) calling_class = calling_class.__class__.__name__ if calling_class else "" calling_function = inspect.stack()[1].function api_response = NSEApiResponse( action = f"{calling_class}.{calling_function}", url = url or self._data_url, method = "GET", inputs = {"headers": headers, "params": params} ) # 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 self._cookies is None: self._cookies = raw_response.cookies else: self._cookies.update(raw_response.cookies) 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