Kite-connection using plain old requests library in python

automatic-login-kiteconnect.py

# -*- 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)

connection_info.json

{
  "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"
}

logging.json

{
  "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"
    ]
  }
}

requirements.txt

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

what is the access_credentials.json file?

I don't seem to be using anything named access_credentials.json. Can you elaborate further?

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 ?

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.