Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save GannyS/619e29dc999ffd014e80ec69ae190b93 to your computer and use it in GitHub Desktop.
Save GannyS/619e29dc999ffd014e80ec69ae190b93 to your computer and use it in GitHub Desktop.
Kite-connection using plain old requests library in python
# -*- coding: utf-8 -*-
import json
import os
import logmatic
import logging
from logging.config import dictConfig
import requests
from http import HTTPStatus
__author__ = "GannyS"
# TODO: These can be passed in as command line arguments. Make this change later!
# Name of file containing configuration information for the logging framework
LOG_FILE_NAME = "logging.json"
# Name of file containing KITE connection information
CONNECTION_INFO_FILE_NAME = "connection_info.json"
# Logger for this module
LOGGER = logging.getLogger(__name__)
def init_logging():
"""
Initialize the logging framework with the configuration
:return: Nothing
:rtype: None
"""
here = os.path.dirname(__file__)
log_fp = os.path.join(here, LOG_FILE_NAME)
if not os.path.exists(log_fp):
raise FileNotFoundError("Logging configuration file not found!")
with open(log_fp, mode="r") as log_config:
dictConfig(config=json.load(log_config))
LOGGER.info("Logging framework initialized!")
def load_kite_config():
"""
Load the kite configuration information from the json file
:return: A python dictionary
:rtype: dict
"""
# Get the directory where this python script is being executed
here = os.path.dirname(__file__)
connect_fp = os.path.join(here, CONNECTION_INFO_FILE_NAME)
if not os.path.exists(connect_fp):
raise FileNotFoundError("Connection information file not found!")
with open(connect_fp, mode="r") as connect_file:
config = json.load(connect_file)
LOGGER.info("Kite connection information loaded successfully from file!")
return config
def kite_prelogin(config, http_session):
"""
Perform pre-login into kite
:param config: The python dictionary containing kite configuration information
:type config: dict
:param http_session: A http session
:type http_session: :class:`requests.Session`
:return: The response referer url
:rtype: str
"""
url = config["LOGIN_REFERER"] + config["KITE_API_KEY"]
response = http_session.get(url=url)
return response.url
def login_kite(config, http_session):
"""
Perform a login
:param config: The python dictionary containing kite configuration information
:type config: dict
:param http_session: A http session
:type http_session: :class:`requests.Session`
:return: The response payload as a python dictionary
:rtype: dict
"""
url = config["LOGIN_URL"]
data = dict()
data["user_id"] = config["USER"]
data["password"] = config["PASSWORD"]
response = http_session.post(url=url, data=data)
# Deserialize the response content
resp_dict = json.loads(response.content)
if "message" in resp_dict.keys():
# Since logging framework already expects message as a parameter change the key
resp_dict["err_message"] = resp_dict["message"]
del resp_dict["message"]
if response.status_code != HTTPStatus.OK:
LOGGER.error("Login failure", extra=resp_dict)
raise ConnectionError("Login failed!")
return resp_dict
def kite_twofa(login_resp, config, http_session):
"""
Perform kite-two-factor authentication
:param login_resp: The response payload from the primary user login
:type login_resp: dict
:param config: The python dictionary containing kite configuration information
:type config: dict
:param http_session: A http session
:type http_session: :class:`requests.Session`
:return: The response payload as a python dictionary
:rtype: dict
"""
url = config["TWOFA_URL"]
data = dict()
data["user_id"] = config["USER"]
data["request_id"] = login_resp["data"]["request_id"]
data["twofa_value"] = str(config["PIN"])
response = http_session.post(url=url, data=data)
# Deserialize the response content
resp_dict = json.loads(response.content)
if "message" in resp_dict.keys():
# Since logging framework already expects message as a parameter change the key
resp_dict["err_message"] = resp_dict["message"]
del resp_dict["message"]
if response.status_code != HTTPStatus.OK:
LOGGER.error("Two-factor authentication failure", extra=resp_dict)
raise ConnectionError("Two-factor authentication failed!")
return resp_dict
if __name__ == "__main__":
# Initialize logging framework
init_logging()
# Load the kite configuration information
kite_config = load_kite_config()
sess = requests.Session()
# Attempt pre-login
ref_url = kite_prelogin(config=kite_config, http_session=sess)
# Attempt a login and get the response as a dictionary
user_pass_login_resp = login_kite(config=kite_config, http_session=sess)
LOGGER.info("Login successful!")
# Attempt two-factor auth
two_fa_resp = kite_twofa(login_resp=user_pass_login_resp, config=kite_config, http_session=sess)
LOGGER.info("Two-factor authentication passed!", extra=two_fa_resp)
{
"KITE_API_KEY": "",
"USER": "",
"PASSWORD": "",
"PIN": "",
"LOGIN_REFERER": "https://kite.trade/connect/login?v=3&api_key=",
"LOGIN_URL": "https://kite.zerodha.com/api/login",
"TWOFA_URL": "https://kite.zerodha.com/api/twofa"
}
{
"version": 1,
"disable_existing_loggers": false,
"formatters": {
"package_formatter": {
"format": "[%(asctime)s] - [%(levelname)s] - [%(name)s] : %(message)s"
},
"json": {
"()": "logmatic.JsonFormatter"
}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "DEBUG",
"formatter": "json"
}
},
"loggers": {
},
"root": {
"level": "INFO",
"handlers": [
"console"
]
}
}
certifi==2020.4.5.1
chardet==3.0.4
idna==2.9
logmatic-python==0.1.7
python-json-logger==0.1.11
requests==2.23.0
urllib3==1.25.8
@Naveenchandra07
Copy link

Hello, thanks for this helpful post. I was able to login successfully using above code. But how can I create kiteconnect ( https://github.com/zerodha/pykiteconnect) object and using it as API trading ?

@rajivgpta
Copy link

I just now created a simple version that returns kiteconnect and works with external 2FA TOTP: https://github.com/rajivgpta/kite-api-autologin . Working as of 23 feb 2023.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment