ow-stats/bnet_retriever/credential.py

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
            }
        })