Spaces:
Paused
Paused
File size: 1,964 Bytes
5b01a63 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 |
import json
import time
from typing import (
Any,
Dict,
Hashable,
Optional,
)
from ..base_client import FrameworkIntegration
class StarletteIntegration(FrameworkIntegration):
async def _get_cache_data(self, key: Hashable):
value = await self.cache.get(key)
if not value:
return None
try:
return json.loads(value)
except (TypeError, ValueError):
return None
async def get_state_data(self, session: Optional[Dict[str, Any]], state: str) -> Dict[str, Any]:
key = f'_state_{self.name}_{state}'
if self.cache:
value = await self._get_cache_data(key)
elif session is not None:
value = session.get(key)
else:
value = None
if value:
return value.get('data')
return None
async def set_state_data(self, session: Optional[Dict[str, Any]], state: str, data: Any):
key = f'_state_{self.name}_{state}'
if self.cache:
await self.cache.set(key, json.dumps({'data': data}), self.expires_in)
elif session is not None:
now = time.time()
session[key] = {'data': data, 'exp': now + self.expires_in}
async def clear_state_data(self, session: Optional[Dict[str, Any]], state: str):
key = f'_state_{self.name}_{state}'
if self.cache:
await self.cache.delete(key)
elif session is not None:
session.pop(key, None)
self._clear_session_state(session)
def update_token(self, token, refresh_token=None, access_token=None):
pass
@staticmethod
def load_config(oauth, name, params):
if not oauth.config:
return {}
rv = {}
for k in params:
conf_key = f'{name}_{k}'.upper()
v = oauth.config.get(conf_key, default=None)
if v is not None:
rv[k] = v
return rv
|