Classes:
| Name | Description |
|---|---|
KotakAPIException |
Raised for Kotak broker API errors. |
KotakAPI |
Kotak API class. |
Bases: BrokerAPIException
Raised for Kotak broker API errors.
Example programs
examples/stock_brokers/api/kotak/KotakAPIException/example_1_catching_a_refused_token.py
"""Catches the `KotakAPIException` that `KotakAPI` raises when Kotak refuses the access token.
When Kotak answers with an HTTP error, `KotakAPI`'s `_request` raises `KotakAPIException` instead of returning an answer. Kotak names the error in `errorCode` and explains it in `message`, and `_request` copies both into the exception.
The real constructor reads the broker's settings from MongoDB, probes `GET /quick/user/positions`, and logs in to Kotak with a TOTP and an MPIN exchanged for a session token and the API host Kotak assigns to that session when the stored token no longer works. This program must never log in, so a small subclass, `KotakAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that already holds the stored login. `requests.request` is replaced, only while each request is sent, by a stand-in server that records the request and answers with a recorded-looking Kotak response, so nothing reaches the network. Here the stand-in server answers with the HTTP 401 Kotak sends for a token it no longer accepts.
Notice that `code` holds Kotak's own error code rather than the HTTP status, so a caller can tell an expired token from other refusals.
Run it from the project root:
python examples/stock_brokers/api/kotak/KotakAPIException/example_1_catching_a_refused_token.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.kotak import (
KotakAPI,
KotakAPIException,
)
STORED_LOGIN = {
'broker_name': 'kotak',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
'sid': 'morning-sid',
'base_url': 'https://e21.kotaksecurities.com',
}
REFUSAL = {
'errorCode': '900901',
'message': 'Invalid Credentials',
}
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the `last_login` hash that every process reads its token from.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
class RecordedBrokerServer:
"""A stand-in for `requests.request` that answers every call with one recorded Kotak response and remembers what was sent.
Attributes:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
calls (list): One dictionary per request, holding its method, URL, headers, body and certificate setting.
"""
def __init__(self, status_code, content_type, body):
"""Holds the answer to give.
Args:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.content_type = content_type
self.body = body
self.calls = []
def request(self, method, url, **options):
"""Records one request and answers it, in place of `requests.request`.
Args:
method (str): The HTTP method.
url (str): The URL requested.
**options (Any): The remaining `requests.request` arguments, such as `headers` and `data`.
Returns:
requests.Response: The recorded answer.
"""
self.calls.append({
'method': method,
'url': url,
'headers': options.get('headers'),
'data': options.get('data'),
'verify': options.get('verify'),
})
response = requests.Response()
response.status_code = self.status_code
response.headers['Content-Type'] = self.content_type
response._content = self.body.encode('utf-8')
response.url = url
return response
class KotakAPIWithoutLogin(KotakAPI):
"""A `KotakAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `kotak` settings document from MongoDB and then logs in to Kotak if the stored token no longer works. This subclass sets the same attributes from what it is given and does nothing else, so every request it makes still runs `KotakAPI`'s own `_request`.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
MongoDB is left out because the stored login is handed in here and is already in the cache, which is the first place `_current_login` looks.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the stored login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'kotak'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('kotak')
self._settings = settings
self._last_login = stored_login
self.login_response = None
class CatchingARefusedTokenExample:
"""Sends a request that Kotak refuses and prints the exception it raises.
Attributes:
server (RecordedBrokerServer): The stand-in for Kotak's API, answering with a refusal.
api (KotakAPIWithoutLogin): The API object being shown.
"""
def __init__(self):
"""Stores an expired login and builds the API object and the stand-in server.
Returns:
None: This method returns nothing.
"""
cache = SharedLoginCache()
cache.hset('last_login', 'kotak', json.dumps(STORED_LOGIN))
settings = {}
self.server = RecordedBrokerServer(401, 'application/json', json.dumps(REFUSAL))
self.api = KotakAPIWithoutLogin(cache, settings, STORED_LOGIN)
def run(self):
"""Sends the request, catches the exception and prints its fields.
Returns:
None: This method returns nothing.
"""
print(f'Kotak answers HTTP {self.server.status_code} with: {self.server.body}')
with unittest.mock.patch('requests.request', self.server.request):
try:
self.api.get(url=self.api.url('/quick/user/positions'))
except KotakAPIException as error:
print(f'Caught: {type(error).__name__}')
print(f'code: {error.code}')
print(f'message: {error.message}')
if __name__ == '__main__':
CatchingARefusedTokenExample().run()
It prints:
examples/stock_brokers/api/kotak/KotakAPIException/example_2_catching_a_gateway_error.py
"""Catches a gateway error from Kotak through the base class that every broker's exception shares.
A load balancer in front of Kotak's API sometimes answers with an HTML page instead of JSON, for example a 502 Bad Gateway while the service behind it restarts. `KotakAPI`'s `_request` cannot read an error code from such a page, so it raises `KotakAPIException` with the HTTP status as `code` and the page's text as `message`.
Every broker's exception class derives from `BrokerAPIException`, so code that talks to several brokers can catch that one class and still read `code` and `message`. This program catches it that way.
The real constructor reads the broker's settings from MongoDB, probes `GET /quick/user/positions`, and logs in to Kotak with a TOTP and an MPIN exchanged for a session token and the API host Kotak assigns to that session when the stored token no longer works. This program must never log in, so a small subclass, `KotakAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that already holds the stored login. `requests.request` is replaced, only while each request is sent, by a stand-in server that records the request and answers with a recorded-looking Kotak response, so nothing reaches the network.
Notice that the exception caught as `BrokerAPIException` is still a `KotakAPIException`, and that its `code` is the integer 502 rather than a code of Kotak's own.
Run it from the project root:
python examples/stock_brokers/api/kotak/KotakAPIException/example_2_catching_a_gateway_error.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.base import (
BrokerAPIException,
)
from stock_brokers.api.kotak import (
KotakAPI,
KotakAPIException,
)
STORED_LOGIN = {
'broker_name': 'kotak',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
'sid': 'morning-sid',
'base_url': 'https://e21.kotaksecurities.com',
}
GATEWAY_PAGE = '<html><head><title>502 Bad Gateway</title></head><body><center><h1>502 Bad Gateway</h1></center></body></html>'
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the `last_login` hash that every process reads its token from.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
class RecordedBrokerServer:
"""A stand-in for `requests.request` that answers every call with one recorded Kotak response and remembers what was sent.
Attributes:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
calls (list): One dictionary per request, holding its method, URL, headers, body and certificate setting.
"""
def __init__(self, status_code, content_type, body):
"""Holds the answer to give.
Args:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.content_type = content_type
self.body = body
self.calls = []
def request(self, method, url, **options):
"""Records one request and answers it, in place of `requests.request`.
Args:
method (str): The HTTP method.
url (str): The URL requested.
**options (Any): The remaining `requests.request` arguments, such as `headers` and `data`.
Returns:
requests.Response: The recorded answer.
"""
self.calls.append({
'method': method,
'url': url,
'headers': options.get('headers'),
'data': options.get('data'),
'verify': options.get('verify'),
})
response = requests.Response()
response.status_code = self.status_code
response.headers['Content-Type'] = self.content_type
response._content = self.body.encode('utf-8')
response.url = url
return response
class KotakAPIWithoutLogin(KotakAPI):
"""A `KotakAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `kotak` settings document from MongoDB and then logs in to Kotak if the stored token no longer works. This subclass sets the same attributes from what it is given and does nothing else, so every request it makes still runs `KotakAPI`'s own `_request`.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
MongoDB is left out because the stored login is handed in here and is already in the cache, which is the first place `_current_login` looks.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the stored login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'kotak'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('kotak')
self._settings = settings
self._last_login = stored_login
self.login_response = None
class CatchingAGatewayErrorExample:
"""Sends a request that a gateway answers with an HTML error page and catches the result as a `BrokerAPIException`.
Attributes:
server (RecordedBrokerServer): The stand-in for Kotak's API, answering with a gateway error.
api (KotakAPIWithoutLogin): The API object being shown.
"""
def __init__(self):
"""Stores a login and builds the API object and the stand-in server.
Returns:
None: This method returns nothing.
"""
cache = SharedLoginCache()
cache.hset('last_login', 'kotak', json.dumps(STORED_LOGIN))
settings = {}
self.server = RecordedBrokerServer(502, 'text/html; charset=utf-8', GATEWAY_PAGE)
self.api = KotakAPIWithoutLogin(cache, settings, STORED_LOGIN)
def run(self):
"""Sends the request, catches the exception through the base class and prints what it holds.
Returns:
None: This method returns nothing.
"""
with unittest.mock.patch('requests.request', self.server.request):
try:
self.api.get(url=self.api.url('/quick/user/positions'))
except BrokerAPIException as error:
print(f'Caught as BrokerAPIException: {type(error).__name__}')
print(f'Is an instance of KotakAPIException: {isinstance(error, KotakAPIException)}')
print(f'code: {error.code!r}')
print(f'message: {error.message}')
if __name__ == '__main__':
CatchingAGatewayErrorExample().run()
It prints:
BrokerAPIException represents an exception condition related to Broker API calls
-code: (string) error code
-message: (string) error message
Example programs
examples/stock_brokers/api/base/BrokerAPIException/example_1_raising_and_reading_an_error.py
"""Raises a `BrokerAPIException` and reads the code and message it carries.
`BrokerAPIException` is what every broker class raises, through its own subclass, when a broker refuses a request. It holds the broker's error `code` and a `message`, both optional, and passes both to `Exception` as well, so they also appear in `args` and in the exception's text.
This program raises one with a Kite-style code and message, as `ZerodhaAPI` does for an expired token, and one with neither, and prints what a handler can read from each. It needs no data store, no network and no broker.
Notice that the code and message are plain attributes, and that an exception raised without them still has both, set to None.
Run it from the project root:
python examples/stock_brokers/api/base/BrokerAPIException/example_1_raising_and_reading_an_error.py
"""
from stock_brokers.api.base import (
BrokerAPIException,
)
class RaisingAndReadingAnErrorExample:
"""Raises two exceptions and prints what each holds.
Attributes:
errors (list): The exceptions to raise, one with a code and message and one without.
"""
def __init__(self):
"""Builds the exceptions to raise.
Returns:
None: This method returns nothing.
"""
self.errors = [
BrokerAPIException(code='TokenException', message='Incorrect `api_key` or `access_token`.'),
BrokerAPIException(),
]
def run(self):
"""Raises each exception, catches it and prints its fields.
Returns:
None: This method returns nothing.
"""
for error_to_raise in self.errors:
try:
raise error_to_raise
except BrokerAPIException as error:
print(f'code: {error.code!r}')
print(f'message: {error.message!r}')
print(f'args: {error.args!r}')
print(f'text: {error}')
print()
if __name__ == '__main__':
RaisingAndReadingAnErrorExample().run()
It prints:
examples/stock_brokers/api/base/BrokerAPIException/example_2_one_handler_for_every_broker.py
"""Handles the errors of several brokers with one `except BrokerAPIException` clause.
Each broker module defines its own exception class, such as `DhanAPIException` or `ZerodhaAPIException`, and every one of them derives from `BrokerAPIException`. Code that works across brokers, such as a script that polls every broker's funds, can therefore catch the base class once and still read the broker's `code` and `message`, and use the exception's class to say which broker refused.
This program raises three brokers' exceptions with the codes and messages those brokers send for an expired token, catches them all with one handler, and prints what the handler sees. It needs no data store, no network and no broker.
Notice that every exception is caught by the same clause, and that each keeps its broker's own code: Dhan's `errorType`, Kite's `error_type` and Noren's `stat`.
Run it from the project root:
python examples/stock_brokers/api/base/BrokerAPIException/example_2_one_handler_for_every_broker.py
"""
from stock_brokers.api.base import (
BrokerAPIException,
)
from stock_brokers.api.dhan import (
DhanAPIException,
)
from stock_brokers.api.flattrade import (
FlattradeAPIException,
)
from stock_brokers.api.zerodha import (
ZerodhaAPIException,
)
class RefusingBroker:
"""A stand-in for one broker's API object whose every request is refused with that broker's exception.
Attributes:
broker_name (str): The broker's name.
refusal (BrokerAPIException): The exception every request raises.
"""
def __init__(self, broker_name, refusal):
"""Holds the broker's name and the refusal.
Args:
broker_name (str): The broker's name.
refusal (BrokerAPIException): The exception every request raises.
Returns:
None: This method returns nothing.
"""
self.broker_name = broker_name
self.refusal = refusal
def get(self, url):
"""Refuses the request.
Args:
url (str): The URL requested.
Returns:
dict: Never returns.
Raises:
BrokerAPIException: Always, the broker's own subclass of it.
"""
raise self.refusal
class OneHandlerExample:
"""Asks every broker for its funds and handles every refusal in one place.
Attributes:
brokers (list): The stand-in brokers to ask.
"""
def __init__(self):
"""Builds the refusing brokers.
Returns:
None: This method returns nothing.
"""
self.brokers = [
RefusingBroker('dhan', DhanAPIException(code='Invalid_Authentication', message='Client ID or user generated access token is invalid or expired.')),
RefusingBroker('zerodha', ZerodhaAPIException(code='TokenException', message='Incorrect `api_key` or `access_token`.')),
RefusingBroker('flattrade', FlattradeAPIException(code='Not_Ok', message='Session Expired : Invalid Session Key')),
]
def run(self):
"""Sends one request to each broker and prints each refusal.
Returns:
None: This method returns nothing.
"""
for broker in self.brokers:
try:
broker.get(url='/funds')
except BrokerAPIException as error:
print(f'{broker.broker_name}: {type(error).__name__} code={error.code} message={error.message}')
if __name__ == '__main__':
OneHandlerExample().run()
It prints:
dhan: DhanAPIException code=Invalid_Authentication message=Client ID or user generated access token is invalid or expired.
zerodha: ZerodhaAPIException code=TokenException message=Incorrect `api_key` or `access_token`.
flattrade: FlattradeAPIException code=Not_Ok message=Session Expired : Invalid Session Key
Source code in stock_brokers/api/base.py
Bases: BrokerAPI
Kotak API class.
Kotak assigns each session its own API host and returns it as baseUrl in the Login Validate
response - e21, e22, e43 and so on. A call made to any other host is refused with
stCode 200032, "Invalid URL. Please verify the 'baseUrl' in the Login Validate API response",
even when the token is perfectly valid. So no host is hard coded: url() reads the one the
login stored and builds every request path on top of it.
The order update feed, bin/kotak/orders/websocket_order_details, does the same thing with the same stored value.
Kotak also has no profile endpoint. The account's profile arrives only in the Login Validate response, so a login here keeps it on login_response and writes it to the Redis key kotak:user:details through _store_profile, in the shape the other brokers' user/details scripts write. _PROFILE_FIELDS names the fields Kotak's Totp_validate documentation describes, rather than filtering the response, so a token Kotak adds to it later can never land in the profile.
Kotak API class
Example programs
examples/stock_brokers/api/kotak/KotakAPI/example_1_reading_positions.py
"""Reads the open positions from Kotak's REST API with a stored session token.
`KotakAPI` is the object every Kotak script talks to the broker through. Its `get`, `post`, `put`, `patch` and `delete` methods all go through the broker's own `_request`, which adds the credentials of the login in force, sends the request, and wraps a successful answer in a dictionary with `status`, `code`, `data` and a `timestamp`.
The real constructor reads the broker's settings from MongoDB, probes `GET /quick/user/positions`, and logs in to Kotak with a TOTP and an MPIN exchanged for a session token and the API host Kotak assigns to that session when the stored token no longer works. This program must never log in, so a small subclass, `KotakAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that already holds the stored login. `requests.request` is replaced, only while each request is sent, by a stand-in server that records the request and answers with a recorded-looking Kotak response, so nothing reaches the network.
Notice the headers the request carried, which were built from the stored login and, where the broker needs them, from the settings, and that `data` holds only the `data` list, because `_request` unwraps the envelope Kotak puts around every answer. The `timestamp` is left out of the output because it is the moment the program ran.
Run it from the project root:
python examples/stock_brokers/api/kotak/KotakAPI/example_1_reading_positions.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.kotak import (
KotakAPI,
)
STORED_LOGIN = {
'broker_name': 'kotak',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
'sid': 'morning-sid',
'base_url': 'https://e21.kotaksecurities.com',
}
RECORDED_ANSWER = {
'stat': 'Ok',
'stCode': 200,
'data': [
{
'exSeg': 'nse_cm',
'trdSym': 'INFY-EQ',
'tok': '1594',
'prod': 'MIS',
'flBuyQty': '10',
'flSellQty': '0',
'buyAmt': '15432.50',
'sellAmt': '0.00',
},
],
}
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the `last_login` hash that every process reads its token from.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
class RecordedBrokerServer:
"""A stand-in for `requests.request` that answers every call with one recorded Kotak response and remembers what was sent.
Attributes:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
calls (list): One dictionary per request, holding its method, URL, headers, body and certificate setting.
"""
def __init__(self, status_code, content_type, body):
"""Holds the answer to give.
Args:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.content_type = content_type
self.body = body
self.calls = []
def request(self, method, url, **options):
"""Records one request and answers it, in place of `requests.request`.
Args:
method (str): The HTTP method.
url (str): The URL requested.
**options (Any): The remaining `requests.request` arguments, such as `headers` and `data`.
Returns:
requests.Response: The recorded answer.
"""
self.calls.append({
'method': method,
'url': url,
'headers': options.get('headers'),
'data': options.get('data'),
'verify': options.get('verify'),
})
response = requests.Response()
response.status_code = self.status_code
response.headers['Content-Type'] = self.content_type
response._content = self.body.encode('utf-8')
response.url = url
return response
class KotakAPIWithoutLogin(KotakAPI):
"""A `KotakAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `kotak` settings document from MongoDB and then logs in to Kotak if the stored token no longer works. This subclass sets the same attributes from what it is given and does nothing else, so every request it makes still runs `KotakAPI`'s own `_request`.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
MongoDB is left out because the stored login is handed in here and is already in the cache, which is the first place `_current_login` looks.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the stored login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'kotak'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('kotak')
self._settings = settings
self._last_login = stored_login
self.login_response = None
class ReadingPositionsExample:
"""Sends one authenticated request through `KotakAPI` and prints what went out and what came back.
Attributes:
cache (SharedLoginCache): The stand-in for Redis, holding the stored login.
server (RecordedBrokerServer): The stand-in for Kotak's API.
api (KotakAPIWithoutLogin): The API object being shown.
"""
def __init__(self):
"""Stores the morning login and builds the API object and the stand-in server.
Returns:
None: This method returns nothing.
"""
self.cache = SharedLoginCache()
self.cache.hset('last_login', 'kotak', json.dumps(STORED_LOGIN))
settings = {}
self.server = RecordedBrokerServer(200, 'application/json', json.dumps(RECORDED_ANSWER))
self.api = KotakAPIWithoutLogin(self.cache, settings, STORED_LOGIN)
def run(self):
"""Sends the request and prints the request and the answer.
Returns:
None: This method returns nothing.
"""
with unittest.mock.patch('requests.request', self.server.request):
answer = self.api.get(url=self.api.url('/quick/user/positions'))
call = self.server.calls[0]
print(f'Request: {call["method"]} {call["url"]}')
print('Headers sent:')
for name, value in call['headers'].items():
print(f' {name}: {value}')
print(f'Status: {answer["status"]}')
print(f'HTTP code: {answer["code"]}')
print('Data:')
print(json.dumps(answer['data'], indent=4))
if __name__ == '__main__':
ReadingPositionsExample().run()
It prints:
Request: GET https://e21.kotaksecurities.com/quick/user/positions
Headers sent:
neo-fin-key: neotradeapi
Auth: morning-token
Sid: morning-sid
Status: success
HTTP code: 200
Data:
[
{
"exSeg": "nse_cm",
"trdSym": "INFY-EQ",
"tok": "1594",
"prod": "MIS",
"flBuyQty": "10",
"flSellQty": "0",
"buyAmt": "15432.50",
"sellAmt": "0.00"
}
]
examples/stock_brokers/api/kotak/KotakAPI/example_2_using_a_newer_login.py
"""Picks up an access token another process obtained, on the very next request.
Every Kotak script builds its own `KotakAPI`, and more than one of them may log in during a day. A login writes the new token to MongoDB and then to the Redis `last_login` hash, and `KotakAPI`'s `_request` reads that hash again before every request, keeping whichever login is newer by its `last_login` time. So a token obtained by any process is used by every other one without restarting anything.
The program sends two requests through one API object. Between them it writes a newer login into the stand-in for Redis, as another process's login would.
The real constructor reads the broker's settings from MongoDB, probes `GET /quick/user/positions`, and logs in to Kotak with a TOTP and an MPIN exchanged for a session token and the API host Kotak assigns to that session when the stored token no longer works. This program must never log in, so a small subclass, `KotakAPIWithoutLogin`, skips that constructor and is handed the settings and a stand-in for Redis that already holds the stored login. `requests.request` is replaced, only while each request is sent, by a stand-in server that records the request and answers with a recorded-looking Kotak response, so nothing reaches the network.
Notice that the second request carries the newer token although the object was never rebuilt.
Run it from the project root:
python examples/stock_brokers/api/kotak/KotakAPI/example_2_using_a_newer_login.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.kotak import (
KotakAPI,
)
STORED_LOGIN = {
'broker_name': 'kotak',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
'sid': 'morning-sid',
'base_url': 'https://e21.kotaksecurities.com',
}
NEWER_LOGIN = {
'broker_name': 'kotak',
'access_token': 'noon-token',
'last_login': '2026-09-30 12:30:41',
'sid': 'noon-sid',
'base_url': 'https://e21.kotaksecurities.com',
}
RECORDED_ANSWER = {
'stat': 'Ok',
'stCode': 200,
'data': [],
}
class SharedLoginCache:
"""A dictionary-backed stand-in for Redis, holding the `last_login` hash that every process reads its token from.
Attributes:
hashes (dict): Each hash name mapped to a dictionary of its fields.
"""
def __init__(self):
"""Starts with nothing stored.
Returns:
None: This method returns nothing.
"""
self.hashes = {}
def hget(self, name, key):
"""Reads one field of a hash.
Args:
name (str): The hash name.
key (str): The field name.
Returns:
str | None: The stored value, or None when there is none.
"""
fields = self.hashes.get(name, {})
return fields.get(key)
def hset(self, name, key, value):
"""Writes one field of a hash, as a login does.
Args:
name (str): The hash name.
key (str): The field name.
value (str): The value to store.
Returns:
int: Always 1, the number of fields Redis reports written.
"""
if name not in self.hashes:
self.hashes[name] = {}
self.hashes[name][key] = value
return 1
class RecordedBrokerServer:
"""A stand-in for `requests.request` that answers every call with one recorded Kotak response and remembers what was sent.
Attributes:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
calls (list): One dictionary per request, holding its method, URL, headers, body and certificate setting.
"""
def __init__(self, status_code, content_type, body):
"""Holds the answer to give.
Args:
status_code (int): The HTTP status every answer carries.
content_type (str): The Content-Type header every answer carries.
body (str): The body every answer carries.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.content_type = content_type
self.body = body
self.calls = []
def request(self, method, url, **options):
"""Records one request and answers it, in place of `requests.request`.
Args:
method (str): The HTTP method.
url (str): The URL requested.
**options (Any): The remaining `requests.request` arguments, such as `headers` and `data`.
Returns:
requests.Response: The recorded answer.
"""
self.calls.append({
'method': method,
'url': url,
'headers': options.get('headers'),
'data': options.get('data'),
'verify': options.get('verify'),
})
response = requests.Response()
response.status_code = self.status_code
response.headers['Content-Type'] = self.content_type
response._content = self.body.encode('utf-8')
response.url = url
return response
class KotakAPIWithoutLogin(KotakAPI):
"""A `KotakAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `kotak` settings document from MongoDB and then logs in to Kotak if the stored token no longer works. This subclass sets the same attributes from what it is given and does nothing else, so every request it makes still runs `KotakAPI`'s own `_request`.
"""
def __init__(self, cache, settings, stored_login):
"""Sets the attributes the real constructor would have loaded.
MongoDB is left out because the stored login is handed in here and is already in the cache, which is the first place `_current_login` looks.
Args:
cache (SharedLoginCache): The stand-in for Redis, already holding the stored login.
settings (dict): The broker's settings document.
stored_login (dict): The broker's `last_login` document, as the real constructor reads it from MongoDB.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'kotak'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('kotak')
self._settings = settings
self._last_login = stored_login
self.login_response = None
class UsingANewerLoginExample:
"""Sends a request, lets another process log in, and sends another through the same object.
Attributes:
cache (SharedLoginCache): The stand-in for Redis, holding the shared login.
server (RecordedBrokerServer): The stand-in for Kotak's API.
api (KotakAPIWithoutLogin): The API object being shown.
"""
def __init__(self):
"""Stores the morning login and builds the API object and the stand-in server.
Returns:
None: This method returns nothing.
"""
self.cache = SharedLoginCache()
self.cache.hset('last_login', 'kotak', json.dumps(STORED_LOGIN))
settings = {}
self.server = RecordedBrokerServer(200, 'application/json', json.dumps(RECORDED_ANSWER))
self.api = KotakAPIWithoutLogin(self.cache, settings, STORED_LOGIN)
def credential(self, call):
"""The credential one recorded request carried.
Args:
call (dict): One request recorded by the stand-in server.
Returns:
str: The `Auth` and `Sid` headers, the session token and session id Kotak issued together.
"""
auth = call['headers']['Auth']
sid = call['headers']['Sid']
return f'Auth={auth} Sid={sid}'
def run(self):
"""Sends a request, writes a newer login, sends another request and prints what each carried.
Returns:
None: This method returns nothing.
"""
with unittest.mock.patch('requests.request', self.server.request):
self.api.get(url=self.api.url('/quick/user/positions'))
print(f'First request carried: {self.credential(self.server.calls[0])}')
self.cache.hset('last_login', 'kotak', json.dumps(NEWER_LOGIN))
print('Another process logged in again at 12:30:41 and wrote its login to Redis.')
with unittest.mock.patch('requests.request', self.server.request):
answer = self.api.get(url=self.api.url('/quick/user/positions'))
print(f'Second request carried: {self.credential(self.server.calls[1])}')
print(f'Request: {self.server.calls[1]["method"]} {self.server.calls[1]["url"]}')
print(f'Status: {answer["status"]}, HTTP code: {answer["code"]}')
print('Data:')
print(json.dumps(answer['data'], indent=4))
if __name__ == '__main__':
UsingANewerLoginExample().run()
It prints:
examples/stock_brokers/api/kotak/KotakAPI/example_3_building_urls_on_the_assigned_host.py
"""Builds Kotak request URLs on the API host that Kotak assigned to the session.
Kotak gives every session its own API host, returned as `baseUrl` when the login is validated, and refuses a call sent to any other host even when the token is good. So `KotakAPI.url` builds every URL on the `base_url` of the stored login rather than on a fixed host. Kotak has sent that host with and without a scheme and with and without a trailing slash, so `url` tidies it before use, and an account that has never logged in, or whose last login failed and stored the string "None", gets the fallback host `https://gw-napi.kotaksecurities.com`.
The real constructor reads the stored login from MongoDB and logs in to Kotak with a TOTP and an MPIN when the stored token no longer works. This program must never log in, so a small subclass, `KotakAPIWithoutLogin`, skips that constructor and is handed the stored login directly. `url` sends nothing, so no stand-in for Redis or the network is needed.
Notice that the same two paths, one written with a leading slash and one without, come out on a clean `https://` host in every case, and that the last two logins fall back to the default host.
Run it from the project root:
python examples/stock_brokers/api/kotak/KotakAPI/example_3_building_urls_on_the_assigned_host.py
"""
import logging
from stock_brokers.api.kotak import (
KotakAPI,
)
class KotakAPIWithoutLogin(KotakAPI):
"""A `KotakAPI` that is handed its stored login instead of loading it and logging in.
The real constructor reads the `kotak` settings and login documents from MongoDB and logs in to Kotak when the stored token no longer works. This subclass sets only what `url` reads.
"""
def __init__(self, stored_login):
"""Sets the attributes the real constructor would have loaded.
Args:
stored_login (dict | None): The broker's `last_login` document, or None for an account that has never logged in.
Returns:
None: This method returns nothing.
"""
self._broker_name = 'kotak'
self._cache = None
self._mongo_db = None
self._logger = logging.getLogger('kotak')
self._settings = {}
self._last_login = stored_login
self.login_response = None
class BuildingUrlsExample:
"""Builds the same two URLs for logins that stored their host in different forms.
Attributes:
stored_logins (list): Pairs of (description, stored login document or None).
paths (list): The API paths to build URLs for.
"""
def __init__(self):
"""Lists the stored logins and the paths.
Returns:
None: This method returns nothing.
"""
self.stored_logins = [
(
'host stored in full',
{
'broker_name': 'kotak',
'access_token': 'session-token',
'sid': 'session-id',
'base_url': 'https://e21.kotaksecurities.com',
},
),
(
'host without a scheme, with a trailing slash',
{
'broker_name': 'kotak',
'access_token': 'session-token',
'sid': 'session-id',
'base_url': 'e43.kotaksecurities.com/',
},
),
(
'host padded with spaces',
{
'broker_name': 'kotak',
'access_token': 'session-token',
'sid': 'session-id',
'base_url': ' https://e22.kotaksecurities.com/ ',
},
),
(
'a failed login that stored "None"',
{
'broker_name': 'kotak',
'access_token': 'None',
'base_url': 'None',
},
),
(
'an account that has never logged in',
None,
),
]
self.paths = [
'/quick/user/limits',
'portfolio/v1/holdings',
]
def run(self):
"""Prints the URLs each stored login produces.
Returns:
None: This method returns nothing.
"""
for description, stored_login in self.stored_logins:
api = KotakAPIWithoutLogin(stored_login)
print(f'{description}:')
for path in self.paths:
print(f' {path} -> {api.url(path)}')
if __name__ == '__main__':
BuildingUrlsExample().run()
It prints:
host stored in full:
/quick/user/limits -> https://e21.kotaksecurities.com/quick/user/limits
portfolio/v1/holdings -> https://e21.kotaksecurities.com/portfolio/v1/holdings
host without a scheme, with a trailing slash:
/quick/user/limits -> https://e43.kotaksecurities.com/quick/user/limits
portfolio/v1/holdings -> https://e43.kotaksecurities.com/portfolio/v1/holdings
host padded with spaces:
/quick/user/limits -> https://e22.kotaksecurities.com/quick/user/limits
portfolio/v1/holdings -> https://e22.kotaksecurities.com/portfolio/v1/holdings
a failed login that stored "None":
/quick/user/limits -> https://gw-napi.kotaksecurities.com/quick/user/limits
portfolio/v1/holdings -> https://gw-napi.kotaksecurities.com/portfolio/v1/holdings
an account that has never logged in:
/quick/user/limits -> https://gw-napi.kotaksecurities.com/quick/user/limits
portfolio/v1/holdings -> https://gw-napi.kotaksecurities.com/portfolio/v1/holdings
Methods:
| Name | Description |
|---|---|
url |
Builds the absolute URL for a Kotak API path, on the host this session was assigned. |
get |
HTTP GET API call to Broker's API server. |
post |
HTTP POST API call to Broker's API server. |
put |
HTTP PUT API call to Broker's API server. |
patch |
HTTP PATCH API call to Broker's API server. |
delete |
HTTP DELETE API call to Broker's API server. |
Source code in stock_brokers/api/kotak.py
72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | |
Builds the absolute URL for a Kotak API path, on the host this session was assigned.
The host is the base_url of the stored login. Kotak has returned it with and without a scheme and with and without a trailing slash, so it is normalized here as well as when the login stores it, rather than trusted. An account that has never logged in has no stored host yet and gets _FALLBACK_BASE_URL.
This stays a separate method because the Kotak scripts in bin/kotak/, bin/check-broker-connections and the Kotak quote source build their request URLs with it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
The path part, with or without a leading slash. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The absolute URL. |
Example usage
From example_2_using_a_newer_login.py
answer = self.api.get(url=self.api.url('/quick/user/positions'))
Source code in stock_brokers/api/kotak.py
Stores the account's profile from a Login Validate response in Redis.
Kotak serves the profile nowhere else. It arrives once, beside the tokens, in the answer to Login Validate, so every process that logs in writes kotak:user:details itself rather than leaving it to the one script that happens to be a session script. The value has the shape the other brokers' user/details scripts write, with timestamp, status, code and the profile under data, and holds only the fields named in _PROFILE_FIELDS.
A failed write is logged and then ignored, because the login itself succeeded and the caller's session is usable without the profile.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
login_data
|
dict
|
The |
required |
Returns:
| Type | Description |
|---|---|
None
|
Nothing is returned; the profile is written to Redis. |
Source code in stock_brokers/api/kotak.py
_request(method, url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
Private generic REST API request to Broker API server. This is specialized into GET, POST, PUT, PATCH and DELETE by public methods.
method: HTTP method to use for the request.url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Source code in stock_brokers/api/kotak.py
192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 | |
The login in force right now, read on every request rather than once at construction.
Any process that logs a broker in writes the new login to MongoDB and then to the Redis
last_login hash. Reading that hash here, on every request, means a token another process
has just obtained is used by this object's very next request, rather than a token loaded at
construction being sent long after it has been replaced.
Redis is the one place a token is read from, and a login the only thing that writes it. Constructors do not copy MongoDB into it, because that would race a login: a process that read MongoDB just before another logged in, and wrote Redis just after, would put the old token back for everyone.
The newer of this object's login and Redis's is used, by their last_login time. The
shared login normally wins. This object's own wins only when it is genuinely newer - a
login this object has just made, or one whose write to Redis failed while its write to
MongoDB, which this object was built from, succeeded. When either time cannot be read,
Redis is trusted.
When Redis holds nothing for the broker, the login is read from MongoDB and written into Redis only if that field is still empty, so it can fill the cache but never overwrite a login that lands in between. When neither store can be read, the object keeps what it has.
Returns:
| Type | Description |
|---|---|
dict | None
|
The |
Source code in stock_brokers/api/base.py
staticmethod
¶
When a login document was made, or None when that cannot be read.
Brokers stamp last_login with local time, some with microseconds and some without.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
document
|
dict | None
|
A |
required |
Returns:
| Type | Description |
|---|---|
datetime | None
|
The moment of the login. |
Source code in stock_brokers/api/base.py
get(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP GET API call to Broker's API server. This queries the Broker's API server for some information. Most responses are JSON encoded. Only a few are CSV encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage
Source code in stock_brokers/api/base.py
post(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP POST API call to Broker's API server. This carries out some action using Broker's API server such as placing an order. Responses are JSON encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage
Source code in stock_brokers/api/base.py
put(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP PUT API call to Broker's API server. This modifies something using Broker's API server such as modifying an order. Responses are JSON encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage
Source code in stock_brokers/api/base.py
patch(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP PATCH API call to Broker's API server. This updates something using Broker's API server such as modifying an order. Responses are JSON encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage
Source code in stock_brokers/api/base.py
delete(url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=None, proxies=None, hooks=None, stream=None, verify=None, cert=None, json=None, verbose=False)
HTTP DELETE API call to Broker's API server. This deletes something using Broker's API server such as cancelling an order or invalidating an access token. Responses are JSON encoded.
url: URL of the API endpoint.params: Dictionary of parameters to be sent as part of the request.data: Dictionary, bytes, or file-like object to send in the body of the request.headers: Dictionary of HTTP headers to be sent with the request.cookies: Dictionary of cookies to be sent with the request.files: Dictionary of files to be sent with the request.auth: Authentication tuple or object to be sent with the request.timeout: Timeout value for the request.allow_redirects: Boolean value indicating whether redirects should be allowed.proxies: Dictionary of proxy settings for the request.hooks: Dictionary of hooks to be called during the request lifecycle.stream: Boolean value indicating whether response content should be streamed.verify: Boolean value indicating whether SSL certificate verification should be performed.cert: Certificate file or tuple containing certificate and key files for client-side SSL certificate verification.json: JSON data to send in the body of the request.verbose: If set to True, the request and response are logged.
Example usage