Merge commit 'c8a95198151bb0dce9119098d0d24c8e86bd5bc9' as 'utils_v2'
This commit is contained in:
@@ -0,0 +1,385 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 27th Dec., 2024
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a way to interface with Zerodha's Kite API in an asynchronous manner. Zerodha provides a great client
|
||||
library, but it works only in sync mode. Here we will try to build an asynchronous version of the same for
|
||||
more advance use cases.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
01. Official documentation: https://kite.trade/docs/connect/v3/
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# System-level activities:
|
||||
import io
|
||||
import os
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import json
|
||||
from utils_v2.system import files
|
||||
from utils_v2.security.hash import Hasher
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
# Data models:
|
||||
from utils_v2.trading.zerodha_kite.models.api_call import ZerodhaKiteApiResponse
|
||||
from utils_v2.trading.zerodha_kite.models.auth_tokens import ZerodhaKiteAuthTokens
|
||||
|
||||
# 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 AsyncZerodhaKite:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
api_key: str,
|
||||
api_secret: str,
|
||||
http_client: httpx.AsyncClient = None,
|
||||
debug = True,
|
||||
debug_prefix = "Z-Kite | ",
|
||||
debug_only_errors = True
|
||||
):
|
||||
|
||||
"""
|
||||
To initialize the instance of this Telegram messenger.
|
||||
:param api_key: The API key that identifies your app. Note that this doesn't change in the lifecycle of the app.
|
||||
:param api_secret: The API secret to access your app. Note that this can be changed from the API portal. This is
|
||||
meant to be kept more secure than the simple API key.
|
||||
:param http_client: An instance of 'httpx' library's AsyncClient. If not given, one will be instantiated
|
||||
internally. It is recommended that, for multi-bot use cases, you provide a common HTTP client from outside.
|
||||
: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 the input configuration:
|
||||
self.__api_key = api_key
|
||||
self.__api_secret = api_secret
|
||||
self.__request_token = None
|
||||
self.__access_token = None
|
||||
self.__checksum = None
|
||||
|
||||
# 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
|
||||
) -> ZerodhaKiteApiResponse:
|
||||
|
||||
"""
|
||||
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 = ZerodhaKiteApiResponse(
|
||||
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
|
||||
|
||||
# Check for errors:
|
||||
await api_response.note_error()
|
||||
|
||||
# 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,
|
||||
params: dict = None,
|
||||
content: str | bytes = None,
|
||||
files: dict = None
|
||||
) -> ZerodhaKiteApiResponse:
|
||||
|
||||
"""
|
||||
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 params: The params to send in the query string itself.
|
||||
:param content: The raw content to be sent in the body (typically as an octet-stream).
|
||||
:param files: Any file that you may want to send.
|
||||
:return: A structured response that includes the raw response, the exception (if any), and so on.
|
||||
"""
|
||||
|
||||
# Prepare the structure of the response:
|
||||
api_response = ZerodhaKiteApiResponse(
|
||||
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,
|
||||
params = params,
|
||||
content = content,
|
||||
files = files
|
||||
)
|
||||
|
||||
# Note down the results:
|
||||
api_response.response = response
|
||||
api_response.httpCode = response.status_code
|
||||
api_response.message = response.reason_phrase
|
||||
|
||||
# Check for errors:
|
||||
await api_response.note_error()
|
||||
|
||||
# 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
|
||||
|
||||
# ┏┓ ┓
|
||||
# ┣┫┓┏╋┣┓
|
||||
# ┛┗┗┻┗┛┗
|
||||
|
||||
@property
|
||||
def login_url(self) -> str:
|
||||
|
||||
"""
|
||||
Generate the login URL that the user can use to log in to his Zerodha account for this app.
|
||||
DOCUMENTATION:
|
||||
01. https://kite.trade/docs/connect/v3/user/
|
||||
:return: The login URL that the user must use.
|
||||
"""
|
||||
|
||||
return f"https://kite.zerodha.com/connect/login?api_key={self.__api_key}"
|
||||
|
||||
def set_request_token(
|
||||
self,
|
||||
request_token: str
|
||||
) -> None:
|
||||
|
||||
"""
|
||||
When the user authorizes the login flow, Zerodha's serve will send you a GET request on the callback URL that
|
||||
you set on the PI portal for your app. This callback will have, among other things, a 'request_token'. The
|
||||
request token is valid only for a very short period, and must be used to get a longer token called
|
||||
'access_token' for actual activities. A checksum is needed for verification. Read about it in the official
|
||||
documentation on Kite's API docs.
|
||||
DOCUMENTATION:
|
||||
01. https://kite.trade/docs/connect/v3/user/
|
||||
:param request_token: The request token received from Zerodha when the user logs in.
|
||||
:return: None.
|
||||
"""
|
||||
|
||||
self.__request_token = request_token
|
||||
hasher = Hasher()
|
||||
hasher.update(self.__api_key + request_token + self.__api_secret)
|
||||
self.__checksum = hasher.hexdigest()
|
||||
|
||||
async def generate_session(
|
||||
self,
|
||||
raise_exception = False
|
||||
) -> ZerodhaKiteAuthTokens | None:
|
||||
|
||||
"""
|
||||
Once we have the 'request_token' from Zerodha's callback, we must generate a session by fetching an access
|
||||
token. The access token will be used to perform most of the actual activities.
|
||||
DOCUMENTATION:
|
||||
01. https://kite.trade/docs/connect/v3/user/
|
||||
:return: Either the auth-token model of Zerodha, or null if the process failed.
|
||||
"""
|
||||
|
||||
# Start by assuming failure:
|
||||
session = None
|
||||
|
||||
# Try to get a session from Zerodha:
|
||||
client_response = await self.__post(
|
||||
url = r"https://api.kite.trade/session/token",
|
||||
headers = {"X-Kite-Version": "3"},
|
||||
data = {
|
||||
"api_key": self.__api_key,
|
||||
"request_token": self.__request_token,
|
||||
"checksum": self.__checksum
|
||||
}
|
||||
)
|
||||
|
||||
# If the API call was successful, we have a valid session:
|
||||
if client_response.success:
|
||||
client_json = await client_response.get_json()
|
||||
|
||||
self.__access_token = client_json["data"]["access_token"]
|
||||
|
||||
print(client_response.to_markdown())
|
||||
print("CLIENT RESPONSE;", json.to_string(await client_response.get_json()))
|
||||
print(json.to_string(client_response.model_dump(), default = str))
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
import asyncio
|
||||
|
||||
async def main():
|
||||
|
||||
# Read the credentials:
|
||||
creds = json.from_file(r"../../../../creds/zerodha/api.json")
|
||||
|
||||
# Create the client:
|
||||
my_kite = AsyncZerodhaKite(
|
||||
api_key = creds["apiKey"],
|
||||
api_secret = creds["apiSecret"]
|
||||
)
|
||||
|
||||
# Login flow:
|
||||
print("LOGIN URL:", my_kite.login_url)
|
||||
my_kite.set_request_token(input("Request Token: "))
|
||||
await my_kite.generate_session()
|
||||
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,177 @@
|
||||
"""
|
||||
|
||||
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
|
||||
@@ -0,0 +1,248 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Saturday, 21st Dec., 2024.
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a data model for describing the tokens to be used for Google's 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, AwareDatetime
|
||||
from typing import Optional, Literal, Union, Dict, List, Any
|
||||
|
||||
# Related to Google:
|
||||
from google.auth.transport.requests import Request
|
||||
from google.oauth2.credentials import Credentials
|
||||
|
||||
# My utils:
|
||||
from utils_v2.string import json
|
||||
from utils_v2.string import regex
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
# To work with date and time:
|
||||
import datetime
|
||||
import dateparser
|
||||
|
||||
# To make API calls:
|
||||
import httpx
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
class ZerodhaKiteAuthTokens(BaseModel):
|
||||
|
||||
userType: str | None = Field(
|
||||
description = "the type of the user",
|
||||
default = None,
|
||||
alias = "user_type",
|
||||
examples = ["individual/ind_with_nom"]
|
||||
)
|
||||
|
||||
userId: str = Field(
|
||||
description = "the id of the user as assigned by zerodha",
|
||||
alias = "user_id",
|
||||
examples = ["ABC123"]
|
||||
)
|
||||
|
||||
email: str | None = Field(
|
||||
description = "the email id of the user",
|
||||
default = None,
|
||||
alias = "email"
|
||||
)
|
||||
|
||||
name: str = Field(
|
||||
description = "the name of the user",
|
||||
alias = "user_name"
|
||||
)
|
||||
|
||||
displayName: str | None = Field(
|
||||
description = "the short display name of the user",
|
||||
default = None,
|
||||
alias = "user_shortname"
|
||||
)
|
||||
|
||||
displayPictureUrl: str | None = Field(
|
||||
description = "the display picture of the user",
|
||||
default = None,
|
||||
alias = "avatar_url"
|
||||
)
|
||||
|
||||
exchanges: List[str] = Field(
|
||||
description = "the list of exchanges this user can trade on",
|
||||
alias = "exchanges"
|
||||
)
|
||||
|
||||
products: List[str] = Field(
|
||||
description = "the list of products (offered by the broker) this user can avail",
|
||||
alias = "products"
|
||||
)
|
||||
|
||||
orderTypes: List[str] = Field(
|
||||
description = "the list of order types this user can place",
|
||||
alias = "order_types"
|
||||
)
|
||||
|
||||
accessToken: str = Field(
|
||||
description = "the main access token to be used in actual requests",
|
||||
alias = "access_token"
|
||||
)
|
||||
|
||||
refreshToken: str | None = Field(
|
||||
description = "token to be used to refresh the access token; may not be provided",
|
||||
default = None,
|
||||
alias = "refresh_token"
|
||||
)
|
||||
|
||||
publicToken: str | None = Field(
|
||||
description = "undocumented on their official documentation",
|
||||
default = None,
|
||||
alias = "public_token"
|
||||
)
|
||||
|
||||
loginTs: AwareDatetime = Field(
|
||||
description = "the time (utc) at which this user logged in",
|
||||
alias = "login_time"
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "ignore"
|
||||
populate_by_name = True
|
||||
|
||||
# ┓┏ ┓• ┓ •
|
||||
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
|
||||
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
|
||||
|
||||
@field_validator("loginTs", mode = "before")
|
||||
def parse_dates(cls, value):
|
||||
if not isinstance(value, datetime.datetime):
|
||||
parsed = date_time.parse_date_time(
|
||||
value,
|
||||
date_formats = ["%Y-%m-%d %H:%M:%S"],
|
||||
timezone = date_time.TIMEZONE_IST
|
||||
)
|
||||
value = parsed if isinstance(parsed, datetime.datetime) else dateparser.parse(value)
|
||||
if isinstance(value, datetime.datetime): value = date_time.to_timezone(value, date_time.TIMEZONE_UTC)
|
||||
return value
|
||||
|
||||
# ┏┓ •
|
||||
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
|
||||
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
|
||||
# ┛
|
||||
|
||||
pass
|
||||
|
||||
# ┏┓ ┏┓
|
||||
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
|
||||
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
|
||||
|
||||
pass
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
sample_token = {
|
||||
"user_type": "individual/ind_with_nom",
|
||||
"email": "khushalpradipsoonderji@gmail.com",
|
||||
"user_name": "Khushal Pradip Soonderji",
|
||||
"user_shortname": "Khushal",
|
||||
"broker": "ZERODHA",
|
||||
"exchanges": [
|
||||
"NSE",
|
||||
"BSE",
|
||||
"NFO",
|
||||
"MF"
|
||||
],
|
||||
"products": [
|
||||
"CNC",
|
||||
"NRML",
|
||||
"MIS",
|
||||
"BO",
|
||||
"CO"
|
||||
],
|
||||
"order_types": [
|
||||
"MARKET",
|
||||
"LIMIT",
|
||||
"SL",
|
||||
"SL-M"
|
||||
],
|
||||
"avatar_url": None,
|
||||
"user_id": "ABC123",
|
||||
"api_key": "same-as-api-key-input-to-this-func",
|
||||
"access_token": "use-this-for-subsequent-activities-like-data-feeds",
|
||||
"public_token": "???",
|
||||
"refresh_token": "",
|
||||
"enctoken": "???",
|
||||
"login_time": "2024-12-20 13:23:20",
|
||||
"meta": {
|
||||
"demat_consent": "consent"
|
||||
}
|
||||
}
|
||||
|
||||
model = ZerodhaKiteAuthTokens(**sample_token)
|
||||
print(json.to_string(model.model_dump(), default = str))
|
||||
Reference in New Issue
Block a user