import pymongo
from pymongo import MongoClient
__version_info__ = ('0', '2', '1')
__version__ = '.'.join(__version_info__)
__author__ = 'Nikita Bobrov'
__license__ = 'MIT/X11'
__copyright__ = '(c) 2017 by Nikita Bobrov'
from datetime import datetime
from uuid import uuid4
from werkzeug.datastructures import CallbackDict
from flask.sessions import SessionInterface, SessionMixin
class MongoSessionManager:
collection_name = 'sessions'
def __init__(self, db='flask_multisession', permanent=True, *args, **kwargs):
self._permanent = permanent
self._client = MongoClient(*args, **kwargs)
self._db = self._client[db]
self._collection = self._db[self.collection_name]
self._check_indexes()
def _check_indexes(self):
sid_index = False
uid_index = False
ttl_index = False
for _, index in enumerate(self._collection.list_indexes()):
if index['key'].get('session_id') is not None:
sid_index = True
if index['key'].get('user_id') is not None:
uid_index = True
if index['key'].get('expired') is not None:
ttl_index = True
if not sid_index:
self._collection.create_index([('session_id', pymongo.HASHED)])
if not uid_index:
self._collection.create_index([('user_id', pymongo.HASHED)], sparse=True)
if not ttl_index:
self._collection.create_index('expired', expireAfterSeconds=0)
def get_session(self, sid):
uid = None
data = {}
if sid:
session = self._collection.find_one({'session_id': sid})
if session is not None:
uid = session.get('user_id', uid)
data = session.get('data', {})
else:
sid = str(uuid4())
session = MongoSession(
data=data,
session_id=sid,
user_id=uid,
permanent=self._permanent,
manager=self
)
return session
def update_session(self, session, expired):
sid = session.session_id
data = {
'session_id': sid,
'expired': expired,
'last_update': datetime.now(),
'data': dict(session)
}
if session.is_authenticated():
data['user_id'] = session.user_id
self._collection.replace_one({'session_id': sid}, data, upsert=True)
def logout_all_devices(self, session):
if session.user_id is not None:
self._collection.update_many(
{'user_id': session.user_id},
{'$unset': {'user_id': ''}}
)
class MongoSession(CallbackDict, SessionMixin):
def __init__(self, data={}, session_id=None, user_id=None, permanent=None, manager=None):
self.manager = manager
self.session_id = session_id
self.user_id = user_id
self.modified = False
self.permanent = permanent
def on_update():
self.modified = True
CallbackDict.__init__(self, data, on_update)
def login(self, uid):
self.user_id = uid
self.modified = True
def logout(self):
self.user_id = None
self.modified = True
def is_authenticated(self):
return self.user_id is not None
def logout_all_devices(self):
self.manager.logout_all_devices(self)
self.logout()
class MongoSessionInterface(SessionInterface):
collection_name = 'sessions'
def __init__(self, *args, **kwargs):
self._manager = MongoSessionManager(*args, **kwargs)
def open_session(self, app, request):
sid = request.cookies.get(app.session_cookie_name)
return self._manager.get_session(sid)
def save_session(self, app, session: MongoSession, response):
domain = self.get_cookie_domain(app)
sid = session.session_id
expired = self.get_expiration_time(app, session)
response.set_cookie(app.session_cookie_name, sid, expires=expired, httponly=True, domain=domain)
if session.modified or expired:
self._manager.update_session(session, expired)
没有合适的资源?快使用搜索试试~ 我知道了~
温馨提示
资源分类:Python库 所属语言:Python 资源全名:Flask-Multi-Session-0.2.1.tar.gz 资源来源:官方 安装方法:https://lanzao.blog.csdn.net/article/details/101784059
资源推荐
资源详情
资源评论
收起资源包目录
Flask-Multi-Session-0.2.1.tar.gz (3个子文件)
Flask-Multi-Session-0.2.1
PKG-INFO 727B
setup.py 1KB
flask_multisession.py 4KB
共 3 条
- 1
资源评论
挣扎的蓝藻
- 粉丝: 14w+
- 资源: 15万+
上传资源 快速赚钱
- 我的内容管理 展开
- 我的资源 快来上传第一个资源
- 我的收益 登录查看自己的收益
- 我的积分 登录查看自己的积分
- 我的C币 登录后查看C币余额
- 我的收藏
- 我的下载
- 下载帮助
最新资源
资源上传下载、课程学习等过程中有任何疑问或建议,欢迎提出宝贵意见哦~我们会及时处理!
点击此处反馈
安全验证
文档复制为VIP权益,开通VIP直接复制
信息提交成功