Add checking the session object for being not None
+
import json
import time
+from typing import (
+ Any,
+ Dict,
+ Hashable,
+ Optional,
+)
+
from ..base_client import FrameworkIntegration
class StartletteIntegration(FrameworkIntegration):
- async def _get_cache_data(self, key):
+ async def _get_cache_data(self, key: Hashable):
value = await self.cache.get(key)
if not value:
return None
@@ -13,29 +21,29 @@ class StartletteIntegration(FrameworkIntegration):
except (TypeError, ValueError):
return None
- async def get_state_data(self, session, state):
+ 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)
- else:
+ elif session is not None:
value = session.get(key)
- if value:
- return value.get('data')
- return None
+ else:
+ value = {}
+ return value.get('data', {})
- async def set_state_data(self, session, state, data):
+ 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, {'data': data}, self.expires_in)
- else:
+ elif session is not None:
now = time.time()
session[key] = {'data': data, 'exp': now + self.expires_in}
- async def clear_state_data(self, session, state):
+ 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)
- else:
+ elif session is not None:
session.pop(key, None)
self._clear_session_state(session)
