Classes:
| Name | Description |
|---|---|
INDMoneyAPIException |
Raised for INDMoney broker API errors. |
INDMoneyAPI |
INDMoneyAPI class |
Bases: BrokerAPIException
Raised for INDMoney broker API errors.
Example programs
examples/stock_brokers/api/indmoney/INDMoneyAPIException/example_1_catching_a_refused_token.py
"""Catches the `INDMoneyAPIException` that `INDMoneyAPI` raises when IND Money refuses the access token.
When IND Money answers with an HTTP error, `INDMoneyAPI`'s `_request` raises `INDMoneyAPIException` instead of returning an answer. IND Money names the error in `error_type`, and `_request` uses that as the exception's `code` and `message` as its message.
The real constructor reads the broker's settings from MongoDB, probes `GET /user/profile`, and logs in to IND Money with an MPIN and a TOTP exchanged for an access token when the stored token no longer works. This program must never log in, so a small subclass, `INDMoneyAPIWithoutLogin`, 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 IND Money response, so nothing reaches the network. Here the stand-in server answers with the HTTP 400 IND Money sends for a token it no longer accepts.
Notice that `code` holds IND Money'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/indmoney/INDMoneyAPIException/example_1_catching_a_refused_token.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.indmoney import (
INDMoneyAPI,
INDMoneyAPIException,
)
STORED_LOGIN = {
'broker_name': 'indmoney',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
}
REFUSAL = {
'status': 'error',
'message': 'Invalid access token',
'error_type': 'RequestValidationException',
}
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 IND Money 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 INDMoneyAPIWithoutLogin(INDMoneyAPI):
"""A `INDMoneyAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `indmoney` settings document from MongoDB and then logs in to IND Money 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 `INDMoneyAPI`'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 = 'indmoney'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('indmoney')
self._settings = settings
self._last_login = stored_login
class CatchingARefusedTokenExample:
"""Sends a request that IND Money refuses and prints the exception it raises.
Attributes:
server (RecordedBrokerServer): The stand-in for IND Money's API, answering with a refusal.
api (INDMoneyAPIWithoutLogin): 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', 'indmoney', json.dumps(STORED_LOGIN))
settings = {}
self.server = RecordedBrokerServer(400, 'application/json', json.dumps(REFUSAL))
self.api = INDMoneyAPIWithoutLogin(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'IND Money answers HTTP {self.server.status_code} with: {self.server.body}')
with unittest.mock.patch('requests.request', self.server.request):
try:
self.api.get(url='https://api.indstocks.com/user/profile')
except INDMoneyAPIException 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/indmoney/INDMoneyAPIException/example_2_catching_a_gateway_error.py
"""Catches a gateway error from IND Money through the base class that every broker's exception shares.
A load balancer in front of IND Money's API sometimes answers with an HTML page instead of JSON, for example a 502 Bad Gateway while the service behind it restarts. `INDMoneyAPI`'s `_request` cannot read an error code from such a page, so it raises `INDMoneyAPIException` 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 /user/profile`, and logs in to IND Money with an MPIN and a TOTP exchanged for an access token when the stored token no longer works. This program must never log in, so a small subclass, `INDMoneyAPIWithoutLogin`, 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 IND Money response, so nothing reaches the network.
Notice that the exception caught as `BrokerAPIException` is still a `INDMoneyAPIException`, and that its `code` is the integer 502 rather than a code of IND Money's own.
Run it from the project root:
python examples/stock_brokers/api/indmoney/INDMoneyAPIException/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.indmoney import (
INDMoneyAPI,
INDMoneyAPIException,
)
STORED_LOGIN = {
'broker_name': 'indmoney',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
}
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 IND Money 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 INDMoneyAPIWithoutLogin(INDMoneyAPI):
"""A `INDMoneyAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `indmoney` settings document from MongoDB and then logs in to IND Money 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 `INDMoneyAPI`'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 = 'indmoney'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('indmoney')
self._settings = settings
self._last_login = stored_login
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 IND Money's API, answering with a gateway error.
api (INDMoneyAPIWithoutLogin): 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', 'indmoney', json.dumps(STORED_LOGIN))
settings = {}
self.server = RecordedBrokerServer(502, 'text/html; charset=utf-8', GATEWAY_PAGE)
self.api = INDMoneyAPIWithoutLogin(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='https://api.indstocks.com/user/profile')
except BrokerAPIException as error:
print(f'Caught as BrokerAPIException: {type(error).__name__}')
print(f'Is an instance of INDMoneyAPIException: {isinstance(error, INDMoneyAPIException)}')
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
INDMoneyAPI class
INDMoneyAPI class
Example programs
examples/stock_brokers/api/indmoney/INDMoneyAPI/example_1_reading_the_profile.py
"""Reads the account profile from IND Money's REST API with a stored access token.
`INDMoneyAPI` is the object every IND Money 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 /user/profile`, and logs in to IND Money with an MPIN and a TOTP exchanged for an access token when the stored token no longer works. This program must never log in, so a small subclass, `INDMoneyAPIWithoutLogin`, 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 IND Money 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` object, because `_request` unwraps the envelope IND Money 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/indmoney/INDMoneyAPI/example_1_reading_the_profile.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.indmoney import (
INDMoneyAPI,
)
STORED_LOGIN = {
'broker_name': 'indmoney',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
}
RECORDED_ANSWER = {
'status': 'success',
'data': {
'user_id': 'IND0000001',
'name': 'DEMO USER',
'segments': [
'EQUITY',
'DERIVATIVE',
],
},
}
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 IND Money 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 INDMoneyAPIWithoutLogin(INDMoneyAPI):
"""A `INDMoneyAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `indmoney` settings document from MongoDB and then logs in to IND Money 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 `INDMoneyAPI`'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 = 'indmoney'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('indmoney')
self._settings = settings
self._last_login = stored_login
class ReadingTheProfileExample:
"""Sends one authenticated request through `INDMoneyAPI` 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 IND Money's API.
api (INDMoneyAPIWithoutLogin): 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', 'indmoney', json.dumps(STORED_LOGIN))
settings = {}
self.server = RecordedBrokerServer(200, 'application/json', json.dumps(RECORDED_ANSWER))
self.api = INDMoneyAPIWithoutLogin(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='https://api.indstocks.com/user/profile')
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__':
ReadingTheProfileExample().run()
It prints:
examples/stock_brokers/api/indmoney/INDMoneyAPI/example_2_using_a_newer_login.py
"""Picks up an access token another process obtained, on the very next request.
Every IND Money script builds its own `INDMoneyAPI`, 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 `INDMoneyAPI`'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 /user/profile`, and logs in to IND Money with an MPIN and a TOTP exchanged for an access token when the stored token no longer works. This program must never log in, so a small subclass, `INDMoneyAPIWithoutLogin`, 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 IND Money 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/indmoney/INDMoneyAPI/example_2_using_a_newer_login.py
"""
import json
import logging
import unittest.mock
import requests
from stock_brokers.api.indmoney import (
INDMoneyAPI,
)
STORED_LOGIN = {
'broker_name': 'indmoney',
'access_token': 'morning-token',
'last_login': '2026-09-30 07:00:05',
}
NEWER_LOGIN = {
'broker_name': 'indmoney',
'access_token': 'noon-token',
'last_login': '2026-09-30 12:30:41',
}
RECORDED_ANSWER = {
'status': 'success',
'data': {
'sod_balance': 113642.0,
'withdrawal_balance': 98310.0,
'pledge_received': 0.0,
},
}
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 IND Money 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 INDMoneyAPIWithoutLogin(INDMoneyAPI):
"""A `INDMoneyAPI` that is handed its settings and cache instead of loading them and logging in.
The real constructor reads the `indmoney` settings document from MongoDB and then logs in to IND Money 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 `INDMoneyAPI`'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 = 'indmoney'
self._cache = cache
self._mongo_db = None
self._logger = logging.getLogger('indmoney')
self._settings = settings
self._last_login = stored_login
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 IND Money's API.
api (INDMoneyAPIWithoutLogin): 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', 'indmoney', json.dumps(STORED_LOGIN))
settings = {}
self.server = RecordedBrokerServer(200, 'application/json', json.dumps(RECORDED_ANSWER))
self.api = INDMoneyAPIWithoutLogin(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 `Authorization` header.
"""
return call['headers']['Authorization']
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='https://api.indstocks.com/funds')
print(f'First request carried: {self.credential(self.server.calls[0])}')
self.cache.hset('last_login', 'indmoney', 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='https://api.indstocks.com/funds')
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:
First request carried: morning-token
Another process logged in again at 12:30:41 and wrote its login to Redis.
Second request carried: noon-token
Request: GET https://api.indstocks.com/funds
Status: success, HTTP code: 200
Data:
{
"sod_balance": 113642.0,
"withdrawal_balance": 98310.0,
"pledge_received": 0.0
}
Methods:
| Name | Description |
|---|---|
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/indmoney.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/indmoney.py
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