import uuid
from sanic_session.base import BaseSessionInterface
from tinydb import Query, TinyDB
__all__ = ['TinyDBSessionInterface']
Session = Query()
class SessionDict(dict):
def __init__(self, **elements):
super().__init__(**elements)
self.modified = False
@property
def sid(self):
return self['sid']
def __setitem__(self, key, value):
self.modified = True
return super().__setitem__(key, value)
def __delitem__(self, key):
self.modified = True
return super().__delitem__(key)
[docs]class TinyDBSessionInterface(BaseSessionInterface):
def __init__(self,
db: TinyDB,
domain: str = None,
expiry: int = 60 * 60 * 24 * 365,
httponly: bool = True,
cookie_name: str = 'session',
table_name: str = 'sessions',
sessioncookie: bool = False):
self.table = db.table(table_name)
self.expiry = expiry
self.cookie_name = cookie_name
self.domain = domain
self.httponly = httponly
self.sessioncookie = sessioncookie
[docs] async def open(self, request) -> SessionDict:
sid = request.cookies.get(self.cookie_name)
if not sid:
sid = uuid.uuid4().hex
session_dict = SessionDict(sid=sid)
else:
data = self.table.get(Session.sid == sid) or dict(sid=sid)
session_dict = SessionDict(**data)
request['session'] = session_dict
return session_dict
[docs] async def save(self, request, response) -> None:
if 'session' not in request or request.method == 'OPTIONS':
return
sid = request['session'].sid
if not request['session']:
self.table.remove(Session.sid == sid)
if request['session'].modified:
self._delete_cookie(request, response)
return
if request['session'].modified:
self.table.upsert(dict(request['session']), Session.sid == sid)
self._set_cookie_expiration(request, response)
[docs] def get_session(self, sid):
return self.table.get(Session.sid == sid)