""" AUTHOR: Khushal P Soonderji DATE: Friday, 27th Dec., 2024. OBJECTIVE: To provide a data model for describing the API response from Zerodha's Kite APIs. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For making data behaviour_models: from pydantic import BaseModel, Field, field_validator, model_validator from typing import Optional, Literal, Union, Dict, List, Any # My utils: from utils_v2.string import json from utils_v2.string import regex # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** class ZerodhaKiteApiResponse(BaseModel): action: str = Field(frozen = True, default = None) url: str = Field(frozen = True) method: str = Field(frozen = True) response: Any = None httpCode: int = None success: bool = False message: str = None data: Any = None error: str = None exception: Any = None # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "forbid" # ┏┓ ┏┓ # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ def to_markdown(self) -> str: """ Use this to summarize the values held in this instance into a markdown-formatted string that can be sent out to admins on chat apps like Telegram. :return: A string in markdown format. """ if self.exception: message = "❌ *ZERODHA (KITE) API EXCEPTION:* ❌\n\n" else: message = "*ZERODHA (KITE) RESPONSE:*\n\n" message += f"*ACTION:*\n`{self.action}`\n\n" message += f"*URL:*\n`{self.url}`\n\n" message += f"*METHOD:*\n`{self.method}`\n\n" message += f"*RESPONSE:*\n`{self.response}`\n\n" message += f"*SUCCESS:*\n`{self.success}`\n\n" message += f"*MESSAGE:*\n`{self.message}`\n\n" message += f"*EXCEPTION:*\n`{self.exception.__class__.__name__}: {str(self.exception)}`\n\n" return message async def get_content(self) -> bytes: """ Get the raw binary content of the body of the response. :return: Raw bytes from the response payload. """ try: return self.response.content except: return b"" async def get_json(self) -> dict | list: """ Get the JSON from the body of the response. :return: A dict or list that represents the JSON payload received in the response. """ try: return self.response.json() except: return {} async def note_error(self) -> None: """ Zerodha has two standard response structures - one for success, one for failure. Refer to their documentation. DOCUMENTATION: 01. https://kite.trade/docs/connect/v3/response-structure/ :return: The error message as a string, or None. """ # If no API was called, we make no changes: if self.response is None: return None # If any API was called: response_json = await self.get_json() response_status = response_json["status"] # If the API call was successful, # we need not worry about errors: if response_status == "success": return None # If the API call failed: self.success = False self.error = response_json["error_type"] self.message = response_json["message"] # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass