session

Establishing a broker session, once, safely, from however many processes want one.

Logging in is a side effect of constructing a broker's API class, which probes an authenticated endpoint and only runs the real login flow when that probe fails. A process that only reads last_login from MongoDB refreshes nothing - one that has been up across the overnight token expiry will keep sending requests against a dead session unless something constructs the API object.

That is what this module is for. The historical candle downloaders call it through BrokerCandles when they log in, and IND Money's instrument ingester calls it when no usable token is stored. Two things make it more than a one line helper.

Only one login at a time. Kite issues one access token per session and a second login invalidates the first. Two processes starting together and both logging in would race, and the loser would come up holding a token that had already been replaced. A Redis lock makes one of them do the work while the other waits for the result.

A floor on how often a login may be attempted. Several brokers log in by driving a headless Chrome and consuming a TOTP. Under Restart=always, a broker outage would otherwise mean a full Selenium login every time the process restarted - a few times a minute, indefinitely, which is how an account gets locked. The rate limiter bounds that regardless of how often the service is restarted, because it lives in Redis rather than in the process.

Classes:

Name	Description
SessionException	Raised when a broker session could not be established.

Functions:

Name	Description
api_class_for	The API class whose construction logs a broker in.
stored_login	A broker's last_login document, or an empty dictionary.
ensure_session	Make sure a usable session exists for a broker, logging in if it does not.

SessionException

Bases: Exception

Raised when a broker session could not be established.

Example programs

A login that failsA login that stores no token

examples/stock_brokers/api/utilities/session/SessionException/example_1_a_login_that_fails.py

"""Catches the `SessionException` that `ensure_session` raises when a broker's login fails.

`ensure_session` is how the candle downloaders and IND Money's instrument download make sure a broker is logged in. It takes a Redis lock so that only one process logs a broker in at a time, records the attempt so that a failing login is not retried more than once every five minutes, and then constructs the broker's API class, whose constructor does the login. Whatever goes wrong inside is re-raised as a `SessionException` naming the broker, with the original error kept as its cause, and the lock is released either way.

Constructing a real broker class would log in to that broker, so this program replaces three things in the session module for the length of the call: `get_cache` returns a small stand-in for Redis, `get_mongo_db` a stand-in for MongoDB with no stored login, and `api_class_for` a stand-in IND Money class whose constructor fails the way a rejected TOTP does. Nothing reaches a data store, the network or a broker.

Notice that the exception's cause is the broker's own error, that the lock is gone afterwards, and that the attempt was recorded but no success was.

Run it from the project root:

    python examples/stock_brokers/api/utilities/session/SessionException/example_1_a_login_that_fails.py
"""

import logging
import unittest.mock

from stock_brokers.api.indmoney import (
    INDMoneyAPIException,
)
from stock_brokers.api.utilities import session as session_module
from stock_brokers.api.utilities.session import (
    SessionException,
)


class StandInRedis:
    """A dictionary-backed stand-in for the Redis commands `ensure_session` uses.

    Attributes:
        keys (dict): Each key mapped to its value.
    """

    def __init__(self):
        """Starts with nothing stored.

        Returns:
            None: This method returns nothing.
        """
        self.keys = {}

    def get(self, name):
        """Reads a key.

        Args:
            name (str): The key.

        Returns:
            str | None: The value, or None when the key is absent.
        """
        return self.keys.get(name)

    def set(self, name, value, nx=False, ex=None):
        """Sets a key, only when it is absent if `nx` is given.

        Args:
            name (str): The key.
            value (str): The value to store.
            nx (bool): True to set the key only when it does not exist.
            ex (int | None): The expiry in seconds, which the stand-in ignores.

        Returns:
            bool: True when the key was set.
        """
        if nx and name in self.keys:
            return False
        self.keys[name] = value
        return True

    def delete(self, name):
        """Removes a key.

        Args:
            name (str): The key.

        Returns:
            int: 1 when the key existed, otherwise 0.
        """
        if name in self.keys:
            del self.keys[name]
            return 1
        return 0


class EmptyCollection:
    """A stand-in for a MongoDB collection that holds no documents."""

    def find_one(self, query, projection=None):
        """Finds nothing.

        Args:
            query (dict): The fields to match.
            projection (dict | None): The fields to return.

        Returns:
            None: There is never a match.
        """
        return None


class EmptyMongo:
    """A stand-in for the MongoDB database in which every collection is empty."""

    def __getitem__(self, name):
        """Returns an empty collection, as `database[name]` does.

        Args:
            name (str): The collection name.

        Returns:
            EmptyCollection: A collection with no documents.
        """
        return EmptyCollection()


class RefusedINDMoneyAPI:
    """A stand-in for `INDMoneyAPI` whose construction, the login, is refused."""

    def __init__(self):
        """Fails the login as IND Money does for a TOTP it rejects.

        Returns:
            None: This method never returns.

        Raises:
            INDMoneyAPIException: Always.
        """
        raise INDMoneyAPIException(code=401, message='Cannot generate access token from INDstocks API: invalid TOTP')


class ALoginThatFailsExample:
    """Asks for an IND Money session whose login fails and prints the resulting exception.

    Attributes:
        cache (StandInRedis): The stand-in for Redis, holding the lock and the attempt markers.
        logger (logging.Logger): Where `ensure_session` reports its progress.
    """

    def __init__(self):
        """Builds the stand-in Redis and the logger.

        Returns:
            None: This method returns nothing.
        """
        self.cache = StandInRedis()
        self.logger = logging.getLogger('indmoney.session')

    def run(self):
        """Calls `ensure_session`, catches the `SessionException` and prints what it and Redis hold.

        Returns:
            None: This method returns nothing.
        """
        with unittest.mock.patch.object(session_module, 'get_cache', return_value=self.cache):
            with unittest.mock.patch.object(session_module, 'get_mongo_db', return_value=EmptyMongo()):
                with unittest.mock.patch.object(session_module, 'api_class_for', return_value=RefusedINDMoneyAPI):
                    try:
                        session_module.ensure_session('indmoney', logger=self.logger)
                    except SessionException as error:
                        print(f'Caught: {type(error).__name__}')
                        print(f'message: {error}')
                        print(f'cause: {type(error.__cause__).__name__} code={error.__cause__.code}')
        print(f'Login lock still held: {"ubi:login:indmoney" in self.cache.keys}')
        print(f'Attempt recorded: {"ubi:login-attempt:indmoney" in self.cache.keys}')
        print(f'Success recorded: {"ubi:login-ok:indmoney" in self.cache.keys}')


if __name__ == '__main__':
    ALoginThatFailsExample().run()

It prints:

Caught: SessionException
message: Could not establish a indmoney session: INDMoneyAPIException: (401, 'Cannot generate access token from INDstocks API: invalid TOTP')
cause: INDMoneyAPIException code=401
Login lock still held: False
Attempt recorded: True
Success recorded: False

examples/stock_brokers/api/utilities/session/SessionException/example_2_a_login_that_stores_no_token.py

"""Catches the `SessionException` that `ensure_session` raises when a login finishes without a usable token.

A broker class's constructor can return without an error and still leave no working token behind, for example when a broker changed its login answer and an earlier failed login had stored the string "None" in place of a token. `ensure_session` does not trust the constructor alone: after it returns, it reads the stored login from MongoDB again and raises a `SessionException` when the token there is missing or "None", so a caller never goes on to send requests with nothing to authenticate them.

Constructing a real broker class would log in to that broker, so this program replaces three things in the session module for the length of the call: `get_cache` returns a small stand-in for Redis, `get_mongo_db` a stand-in for MongoDB whose stored IND Money login holds "None", and `api_class_for` a stand-in IND Money class whose constructor returns quietly without storing anything. Nothing reaches a data store, the network or a broker.

Notice that this exception has no cause, because nothing inside raised, and that the lock was released before the stored login was checked.

Run it from the project root:

    python examples/stock_brokers/api/utilities/session/SessionException/example_2_a_login_that_stores_no_token.py
"""

import logging
import unittest.mock

from stock_brokers.api.utilities import session as session_module
from stock_brokers.api.utilities.session import (
    SessionException,
)


class StandInRedis:
    """A dictionary-backed stand-in for the Redis commands `ensure_session` uses.

    Attributes:
        keys (dict): Each key mapped to its value.
    """

    def __init__(self):
        """Starts with nothing stored.

        Returns:
            None: This method returns nothing.
        """
        self.keys = {}

    def get(self, name):
        """Reads a key.

        Args:
            name (str): The key.

        Returns:
            str | None: The value, or None when the key is absent.
        """
        return self.keys.get(name)

    def set(self, name, value, nx=False, ex=None):
        """Sets a key, only when it is absent if `nx` is given.

        Args:
            name (str): The key.
            value (str): The value to store.
            nx (bool): True to set the key only when it does not exist.
            ex (int | None): The expiry in seconds, which the stand-in ignores.

        Returns:
            bool: True when the key was set.
        """
        if nx and name in self.keys:
            return False
        self.keys[name] = value
        return True

    def delete(self, name):
        """Removes a key.

        Args:
            name (str): The key.

        Returns:
            int: 1 when the key existed, otherwise 0.
        """
        if name in self.keys:
            del self.keys[name]
            return 1
        return 0


class StoredLoginCollection:
    """A stand-in for the MongoDB `last_login` collection, holding one document.

    Attributes:
        document (dict): The stored login.
    """

    def __init__(self, document):
        """Holds the stored login.

        Args:
            document (dict): The stored login.

        Returns:
            None: This method returns nothing.
        """
        self.document = document

    def find_one(self, query, projection=None):
        """Returns the stored login when the query names its broker.

        Args:
            query (dict): The fields to match, here `broker_name`.
            projection (dict | None): The fields to return, which the stand-in ignores.

        Returns:
            dict | None: A copy of the stored login, or None for another broker.
        """
        if query.get('broker_name') != self.document['broker_name']:
            return None
        return dict(self.document)


class StandInMongo:
    """A stand-in for the MongoDB database, holding only the `last_login` collection.

    Attributes:
        last_login (StoredLoginCollection): The `last_login` collection.
    """

    def __init__(self, last_login):
        """Holds the collection.

        Args:
            last_login (StoredLoginCollection): The `last_login` collection.

        Returns:
            None: This method returns nothing.
        """
        self.last_login = last_login

    def __getitem__(self, name):
        """Returns the `last_login` collection, as `database['last_login']` does.

        Args:
            name (str): The collection name, always `last_login` here.

        Returns:
            StoredLoginCollection: The collection.
        """
        return self.last_login


class QuietINDMoneyAPI:
    """A stand-in for `INDMoneyAPI` whose construction returns without storing a token."""

    def __init__(self):
        """Returns without doing anything, as a login that silently stored nothing would.

        Returns:
            None: This method returns nothing.
        """


class ALoginThatStoresNoTokenExample:
    """Asks for an IND Money session whose login leaves no usable token and prints the resulting exception.

    Attributes:
        cache (StandInRedis): The stand-in for Redis, holding the lock and the attempt markers.
        mongo_db (StandInMongo): The stand-in for MongoDB, holding a failed login.
        logger (logging.Logger): Where `ensure_session` reports its progress.
    """

    def __init__(self):
        """Builds the stand-in data stores and the logger.

        Returns:
            None: This method returns nothing.
        """
        self.cache = StandInRedis()
        self.mongo_db = StandInMongo(StoredLoginCollection({
            'broker_name': 'indmoney',
            'access_token': 'None',
            'last_login': '2026-09-29 07:00:04',
        }))
        self.logger = logging.getLogger('indmoney.session')

    def run(self):
        """Calls `ensure_session`, catches the `SessionException` and prints what it and Redis hold.

        Returns:
            None: This method returns nothing.
        """
        with unittest.mock.patch.object(session_module, 'get_cache', return_value=self.cache):
            with unittest.mock.patch.object(session_module, 'get_mongo_db', return_value=self.mongo_db):
                with unittest.mock.patch.object(session_module, 'api_class_for', return_value=QuietINDMoneyAPI):
                    try:
                        session_module.ensure_session('indmoney', logger=self.logger)
                    except SessionException as error:
                        print(f'Caught: {type(error).__name__}')
                        print(f'message: {error}')
                        print(f'cause: {error.__cause__}')
        print(f'Login lock still held: {"ubi:login:indmoney" in self.cache.keys}')


if __name__ == '__main__':
    ALoginThatStoresNoTokenExample().run()

It prints:

Caught: SessionException
message: indmoney login completed but stored no usable access token.
cause: None
Login lock still held: False

_lock_key

_lock_key(broker_name)

Redis key held while one process logs a broker in.

• broker_name is the name of the broker.

Source code in stock_brokers/api/utilities/session.py

def _lock_key(broker_name):
    """
    Redis key held while one process logs a broker in.

    - `broker_name` is the name of the broker.
    """
    return f"ubi:login:{broker_name}"

_attempt_key

_attempt_key(broker_name)

Redis key recording when a broker was last sent through its login flow.

• broker_name is the name of the broker.

Source code in stock_brokers/api/utilities/session.py

def _attempt_key(broker_name):
    """
    Redis key recording when a broker was last sent through its login flow.

    - `broker_name` is the name of the broker.
    """
    return f"ubi:login-attempt:{broker_name}"

_success_key

_success_key(broker_name)

Redis key recording when a broker's session was last confirmed working.

Distinct from the attempt key because the usual outcome of ensure_session is that the stored token was already valid and only a probe happened, which changes nothing in MongoDB. Without a separate record there is no way to tell that from an attempt that failed.

• broker_name is the name of the broker.

Source code in stock_brokers/api/utilities/session.py

def _success_key(broker_name):
    """
    Redis key recording when a broker's session was last confirmed working.

    Distinct from the attempt key because the usual outcome of `ensure_session` is that the
    stored token was already valid and only a probe happened, which changes nothing in MongoDB.
    Without a separate record there is no way to tell that from an attempt that failed.

    - `broker_name` is the name of the broker.
    """
    return f"ubi:login-ok:{broker_name}"

api_class_for

api_class_for(broker_name)

The API class whose construction logs a broker in.

Imported lazily and one at a time: several of these modules import Selenium at module level, and a process that only needs Zerodha should not pay for the other nine.

• broker_name is the name of the broker.

Source code in stock_brokers/api/utilities/session.py

def api_class_for(broker_name):
    """
    The API class whose construction logs a broker in.

    Imported lazily and one at a time: several of these modules import Selenium at module level,
    and a process that only needs Zerodha should not pay for the other nine.

    - `broker_name` is the name of the broker.
    """
    import importlib

    modules = {
        "zerodha": ("stock_brokers.api.zerodha", "ZerodhaAPI"),
        "dhan": ("stock_brokers.api.dhan", "DhanAPI"),
        "flattrade": ("stock_brokers.api.flattrade", "FlattradeAPI"),
        "shoonya": ("stock_brokers.api.shoonya", "ShoonyaAPI"),
        "fyers": ("stock_brokers.api.fyers", "FyersAPI"),
        "groww": ("stock_brokers.api.groww", "GrowwAPI"),
        "kotak": ("stock_brokers.api.kotak", "KotakAPI"),
        "indmoney": ("stock_brokers.api.indmoney", "INDMoneyAPI"),
        "wisdom_capital": ("stock_brokers.api.wisdom_capital", "WisdomCapitalAPI"),
        "stoxkart": ("stock_brokers.api.stoxkart", "StoxkartAPI"),
    }
    if broker_name not in modules:
        raise ValueError(f"No API class known for broker {broker_name}. "
                         f"Known: {', '.join(sorted(modules))}")
    module_name, class_name = modules[broker_name]
    return getattr(importlib.import_module(module_name), class_name)

stored_login

stored_login(broker_name)

A broker's last_login document, or an empty dictionary.

• broker_name is the name of the broker.

Source code in stock_brokers/api/utilities/session.py

def stored_login(broker_name):
    """
    A broker's `last_login` document, or an empty dictionary.

    - `broker_name` is the name of the broker.
    """
    return get_mongo_db()['last_login'].find_one({'broker_name': broker_name}, {'_id': 0}) or {}

_login_timestamp

_login_timestamp(document)

The moment a stored login was made, as an epoch, or None when it cannot be read.

• document is a last_login document.

Source code in stock_brokers/api/utilities/session.py

def _login_timestamp(document):
    """
    The moment a stored login was made, as an epoch, or None when it cannot be read.

    - `document` is a `last_login` document.
    """
    raw = (document or {}).get('last_login')
    if not raw:
        return None
    try:
        return datetime.strptime(raw, "%Y-%m-%d %H:%M:%S.%f").timestamp()
    except (TypeError, ValueError):
        return None

_clear_dead_holder

_clear_dead_holder(broker_name, cache, logger)

Release the login lock if the process holding it no longer exists.

The lock has an expiry, so a crashed holder always clears eventually - but eventually is up to five minutes, and this is reached on the ordinary path of restarting a service while it is inside ensure_session, before its signal handlers are installed. Waiting out the expiry every time would make a restart look like a hang.

The lock holds the holder's process id, and everything that takes it runs on this machine, so a liveness check is a signal 0. Returns whether the lock was cleared.

• broker_name is the name of the broker.
• cache is the Redis client.
• logger is where the release is reported.

Source code in stock_brokers/api/utilities/session.py

def _clear_dead_holder(broker_name, cache, logger):
    """
    Release the login lock if the process holding it no longer exists.

    The lock has an expiry, so a crashed holder always clears eventually - but eventually is up
    to five minutes, and this is reached on the ordinary path of restarting a service while it is
    inside `ensure_session`, before its signal handlers are installed. Waiting out the expiry
    every time would make a restart look like a hang.

    The lock holds the holder's process id, and everything that takes it runs on this machine, so
    a liveness check is a signal 0. Returns whether the lock was cleared.

    - `broker_name` is the name of the broker.
    - `cache` is the Redis client.
    - `logger` is where the release is reported.
    """
    holder = cache.get(_lock_key(broker_name))
    if not holder:
        return False

    try:
        pid = int(holder)
    except (TypeError, ValueError):
        return False

    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        logger.warning(f"The process that held the {broker_name} login lock (pid {pid}) is gone. "
                       f"Releasing it.")
        # Only delete if it is still the same holder, so a lock taken between the read and here
        # is not stolen from a live process.
        if cache.get(_lock_key(broker_name)) == holder:
            cache.delete(_lock_key(broker_name))
            return True
    except PermissionError:
        # The pid exists and belongs to somebody else, so it is alive as far as this matters.
        return False
    return False

ensure_session

ensure_session(broker_name, force=False, logger=None)

Make sure a usable session exists for a broker, logging in if it does not.

Constructing the broker's API class is what does the work: it probes an authenticated endpoint first, so the common case where the stored token is still good costs one HTTP request rather than a login.

Returns the last_login document in force after the call.

• broker_name is the name of the broker.
• force skips the rate limiter, for a deliberate manual re-login.
• logger is where progress is reported. A named logger is made when not supplied.

Source code in stock_brokers/api/utilities/session.py

def ensure_session(broker_name, force=False, logger=None):
    """
    Make sure a usable session exists for a broker, logging in if it does not.

    Constructing the broker's API class is what does the work: it probes an authenticated
    endpoint first, so the common case where the stored token is still good costs one HTTP
    request rather than a login.

    Returns the `last_login` document in force after the call.

    - `broker_name` is the name of the broker.
    - `force` skips the rate limiter, for a deliberate manual re-login.
    - `logger` is where progress is reported. A named logger is made when not supplied.
    """
    logger = logger or get_logger(f"{broker_name}.session")
    cache = get_cache()
    started_at = time.time()

    acquired = cache.set(_lock_key(broker_name), str(os.getpid()),
                         nx=True, ex=LOCK_TIMEOUT_SECONDS)

    if not acquired and _clear_dead_holder(broker_name, cache, logger):
        acquired = cache.set(_lock_key(broker_name), str(os.getpid()),
                             nx=True, ex=LOCK_TIMEOUT_SECONDS)

    if not acquired:
        logger.info(f"Another process is logging {broker_name} in. Waiting for it.")
        return _wait_for_other_process(broker_name, cache, started_at, logger)

    try:
        last_attempt = float(cache.get(_attempt_key(broker_name)) or 0)
        last_success = float(cache.get(_success_key(broker_name)) or 0)
        document = stored_login(broker_name)
        token = document.get('access_token')
        has_token = bool(token) and token != 'None'

        # A session confirmed working within the last interval is simply used. The confirmation
        # has to be recorded separately from the login itself, because the usual outcome is that
        # the stored token was already good and the API class only probed it - which writes
        # nothing to last_login. Comparing against the stored login time instead would treat a
        # perfectly valid token issued this morning as stale.
        #
        # This is the ordinary path: the login unit runs, then the services that need the session
        # start behind it and must not queue up behind a rate limiter meant for failed logins.
        if has_token and not force and last_success:
            age = time.time() - last_success
            if age < MIN_LOGIN_INTERVAL_SECONDS:
                logger.info(f"{broker_name} session was confirmed {age:.0f}s ago. Using it.")
                return document

        # Throttle only a genuine retry: an attempt that was made recently and did not end in a
        # confirmed session. Without this, a broker outage plus Restart=always would mean a full
        # Selenium login every eighty seconds, which is how an account gets locked.
        if last_attempt and not force and last_success < last_attempt:
            age = time.time() - last_attempt
            if age < MIN_LOGIN_INTERVAL_SECONDS:
                wait = MIN_LOGIN_INTERVAL_SECONDS - age
                logger.warning(
                    f"The last {broker_name} login attempt was {age:.0f}s ago and did not "
                    f"produce a working session. Waiting {wait:.0f}s before trying again.")
                time.sleep(wait)

        cache.set(_attempt_key(broker_name), str(time.time()))
        logger.info(f"Establishing a {broker_name} session.")
        api_class_for(broker_name)()
        cache.set(_success_key(broker_name), str(time.time()))
    except Exception as exception:
        raise SessionException(
            f"Could not establish a {broker_name} session: "
            f"{type(exception).__name__}: {exception}") from exception
    finally:
        cache.delete(_lock_key(broker_name))

    document = stored_login(broker_name)
    if not document.get('access_token') or document.get('access_token') == 'None':
        raise SessionException(
            f"{broker_name} login completed but stored no usable access token.")
    logger.info(f"{broker_name} session ready, logged in at {document.get('last_login')}.")
    return document

_wait_for_other_process

_wait_for_other_process(broker_name, cache, started_at, logger)

Wait for whichever process holds the lock to finish, then use its result.

Succeeds as soon as the stored login is newer than this call began, or as soon as the lock is gone and a usable token is present - the second case covers a login that was already valid and so wrote nothing new.

• broker_name is the name of the broker.
• cache is the Redis client.
• started_at is the epoch at which this call began.
• logger is where progress is reported.

Source code in stock_brokers/api/utilities/session.py

def _wait_for_other_process(broker_name, cache, started_at, logger):
    """
    Wait for whichever process holds the lock to finish, then use its result.

    Succeeds as soon as the stored login is newer than this call began, or as soon as the lock is
    gone and a usable token is present - the second case covers a login that was already valid
    and so wrote nothing new.

    - `broker_name` is the name of the broker.
    - `cache` is the Redis client.
    - `started_at` is the epoch at which this call began.
    - `logger` is where progress is reported.
    """
    deadline = started_at + WAIT_FOR_OTHER_SECONDS
    while time.time() < deadline:
        time.sleep(2)
        document = stored_login(broker_name)
        timestamp = _login_timestamp(document)
        if timestamp and timestamp >= started_at:
            logger.info(f"{broker_name} logged in by another process.")
            return document
        if not cache.exists(_lock_key(broker_name)):
            token = document.get('access_token')
            if token and token != 'None':
                logger.info(f"{broker_name} session already in place.")
                return document
            raise SessionException(
                f"The process logging {broker_name} in finished without storing a token.")
    raise SessionException(
        f"Timed out after {WAIT_FOR_OTHER_SECONDS}s waiting for another process to log "
        f"{broker_name} in.")