import time import logging from typing import Dict, Optional from .errors import LoginFailed from ._plugin import Plugin from .utils import serialize from .utils.login import login from .utils.db import Mongo logger = logging.getLogger(__name__) class CredentialManager(Plugin): def __init__(self, runner): super(CredentialManager, self).__init__(runner) self._bnet_user_cred: Optional[Dict] = None @property def raw_cred(self) -> Dict: return serialize.fromstr(Mongo.db.user.find_one({'_id': self._id})['credential']) @property def bnet_user_cred(self) -> str: # Here, self.last_updated_at records if the bnet_user_cred outdated ( 1 hr expiration) # Different from the lastUpdatedAt field in mongodb if self._bnet_user_cred is None or time.time() - self.last_updated_at > 60*55: try: bnet_user_cred, new_cred = login(self.raw_cred) except LoginFailed: logger.info('[%s] login failed') return self._bnet_user_cred = bnet_user_cred self.update_raw_cred(new_cred) logger.info('[%s] Updated credential', self._id) self.last_updated_at = time.time() return self._bnet_user_cred def update_raw_cred(self, new_cred: Dict) -> None: Mongo.db.user.update_one({'_id': self._id}, { '$set': { 'credential': serialize.tostr(new_cred), }, '$currentDate': { 'lastUpdatedAt': True } })