codekingpro/portable-devtools

1import json2import time3 4 5class FrameworkIntegration:6    expires_in = 36007 8    def __init__(self, name, cache=None):9        self.name = name10        self.cache = cache11 12    def _get_cache_data(self, key):13        value = self.cache.get(key)14        if not value:15            return None16        try:17            return json.loads(value)18        except (TypeError, ValueError):19            return None20 21    def _clear_session_state(self, session):22        now = time.time()23        for key in dict(session):24            if '_authlib_' in key:25                # TODO: remove in future26                session.pop(key)27            elif key.startswith('_state_'):28                value = session[key]29                exp = value.get('exp')30                if not exp or exp < now:31                    session.pop(key)32 33    def get_state_data(self, session, state):34        key = f'_state_{self.name}_{state}'35        if self.cache:36            value = self._get_cache_data(key)37        else:38            value = session.get(key)39        if value:40            return value.get('data')41        return None42 43    def set_state_data(self, session, state, data):44        key = f'_state_{self.name}_{state}'45        if self.cache:46            self.cache.set(key, json.dumps({'data': data}), self.expires_in)47        else:48            now = time.time()49            session[key] = {'data': data, 'exp': now + self.expires_in}50 51    def clear_state_data(self, session, state):52        key = f'_state_{self.name}_{state}'53        if self.cache:54            self.cache.delete(key)55        else:56            session.pop(key, None)57            self._clear_session_state(session)58 59    def update_token(self, token, refresh_token=None, access_token=None):60        raise NotImplementedError()61 62    @staticmethod63    def load_config(oauth, name, params):64        raise NotImplementedError()65