Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_session_storage_get():
session = SessionStorage()
with pytest.raises(NotImplementedError):
session.get('s')
with pytest.raises(NotImplementedError):
session['s']
def test_session_storage_delete():
session = SessionStorage()
with pytest.raises(NotImplementedError):
session.delete('s')
with pytest.raises(NotImplementedError):
del session['s']
# -*- coding: utf-8 -*-
from werobot.session import SessionStorage
from werobot.utils import json_loads, json_dumps
class MongoDBStorage(SessionStorage):
"""
MongoDBStorage 会把你的 Session 数据储存在一个 MongoDB Collection 中 ::
import pymongo
import werobot
from werobot.session.mongodbstorage import MongoDBStorage
collection = pymongo.MongoClient()["wechat"]["session"]
session_storage = MongoDBStorage(collection)
robot = werobot.WeRoBot(token="token", enable_session=True,
session_storage=session_storage)
你需要安装 ``pymongo`` 才能使用 MongoDBStorage 。
:param collection: 一个 MongoDB Collection。
# -*- coding: utf-8 -*-
from werobot.session import SessionStorage
from werobot.utils import json_loads, json_dumps
class MongoDBStorage(SessionStorage):
"""
MongoDBStorage 会把你的 Session 数据储存在一个 MongoDB Collection 中 ::
import pymongo
import werobot
from werobot.session.mongodbstorage import MongoDBStorage
collection = pymongo.MongoClient()["wechat"]["session"]
session_storage = MongoDBStorage(collection)
robot = werobot.WeRoBot(token="token", enable_session=True,
session_storage=session_storage)
你需要安装 ``pymongo`` 才能使用 MongoDBStorage 。
:param collection: 一个 MongoDB Collection。
# -*- coding: utf-8 -*-
try:
import anydbm as dbm
assert dbm
except ImportError:
import dbm
from werobot.session import SessionStorage
from werobot.utils import json_loads, json_dumps, to_binary
class FileStorage(SessionStorage):
"""
FileStorage 会把你的 Session 数据以 dbm 形式储存在文件中。
:param filename: 文件名, 默认为 ``werobot_session``
"""
def __init__(self, filename='werobot_session'):
try:
self.db = dbm.open(filename, "c")
except TypeError: # pragma: no cover
# dbm in PyPy requires filename to be binary
self.db = dbm.open(to_binary(filename), "c")
def get(self, id):
"""
根据 id 获取数据。
# -*- coding: utf-8 -*-
from werobot.session import SessionStorage
from werobot.utils import json_loads, json_dumps
class RedisStorage(SessionStorage):
"""
RedisStorage 会把你的 Session 数据储存在 Redis 中 ::
import redis
import werobot
from werobot.session.redisstorage import RedisStorage
db = redis.Redis()
session_storage = RedisStorage(db, prefix="my_prefix_")
robot = werobot.WeRoBot(token="token", enable_session=True,
session_storage=session_storage)
你需要安装 ``redis`` 才能使用 RedisStorage 。
:param redis: 一个 Redis Client。
try:
import anydbm as dbm
assert dbm
except ImportError:
import dbm
import pickle
import logging
from redis import StrictRedis
from werobot.session import SessionStorage
from coolcantonese.util import to_string_type
logger = logging.getLogger(__name__)
class Session(SessionStorage):
"""extend for werobot SessionStorage"""
def __init__(self):
super(Session, self).__init__()
def expire(self, key, expire_seconds):
raise NotImplementedError()
def exists(self, key):
raise NotImplementedError()
class RedisSession(StrictRedis, Session):
def get(self, name):
pickled_value = super(RedisSession, self).get(name)
if pickled_value is None:
# -*- coding: utf-8 -*-
try:
import anydbm as dbm
assert dbm
except ImportError:
import dbm
from werobot.session import SessionStorage
from werobot.utils import json_loads, json_dumps
class FileStorage(SessionStorage):
"""
FileStorage 会把你的 Session 数据以 dbm 形式储存在文件中。
:param filename: 文件名, 默认为 ``werobot_session``
"""
def __init__(self, filename='werobot_session'):
self.db = dbm.open(filename, "c")
def get(self, id):
session_json = self.db.get(id, "{}")
return json_loads(session_json)
def set(self, id, value):
self.db[id] = json_dumps(value)
def delete(self, id):
# -*- coding: utf-8 -*-
from werobot.session import SessionStorage
class MemoryStorage(SessionStorage):
"""
MemoryStorage 会把你的 Session 数据存到内存当中
"""
data_dict = None
def __init__(self):
self.data_dict = {}
def get(self, id):
return self.data_dict.get(id, {})
def set(self, id, value):
self.data_dict[id] = value
def delete(self, id):
del self.data_dict[id]
"""
@version: ??
@author: liangliangyy
@license: MIT Licence
@contact: liangliangyy@gmail.com
@site: https://www.lylinux.net/
@software: PyCharm
@file: MemcacheStorage.py
@time: 2017/8/27 上午2:42
"""
from werobot.session import SessionStorage
from werobot.utils import json_loads, json_dumps
from DjangoBlog.utils import cache
class MemcacheStorage(SessionStorage):
def __init__(self, prefix='ws_'):
self.prefix = prefix
self.cache = cache
@property
def is_available(self):
value = "1"
self.set('checkavaliable', value=value)
return value == self.get('checkavaliable')
def key_name(self, s):
return '{prefix}{s}'.format(prefix=self.prefix, s=s)
def get(self, id):
id = self.key_name(id)
session_json = self.cache.get(id) or '{}'