How to use the werobot.session.SessionStorage function in WeRoBot

To help you get started, we’ve selected a few WeRoBot examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github offu / WeRoBot / tests / test_session.py View on Github external
def test_session_storage_get():
    session = SessionStorage()
    with pytest.raises(NotImplementedError):
        session.get('s')
    with pytest.raises(NotImplementedError):
        session['s']
github offu / WeRoBot / tests / test_session.py View on Github external
def test_session_storage_delete():
    session = SessionStorage()
    with pytest.raises(NotImplementedError):
        session.delete('s')

    with pytest.raises(NotImplementedError):
        del session['s']
github offu / WeRoBot / werobot / session / mongodbstorage.py View on Github external
# -*- 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。
github JoneXiong / oejia_wx / ext_libs / werobot / session / mongodbstorage.py View on Github external
# -*- 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。
github offu / WeRoBot / werobot / session / filestorage.py View on Github external
# -*- 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 获取数据。
github offu / WeRoBot / werobot / session / redisstorage.py View on Github external
# -*- 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。
github kk17 / CoolCantonese / coolcantonese / session.py View on Github external
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:
github JoneXiong / oejia_wx / ext_libs / werobot / session / filestorage.py View on Github external
# -*- 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):
github JoneXiong / oejia_wx / ext_libs / werobot / session / memorystorage.py View on Github external
# -*- 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]
github liangliangyy / DjangoBlog / servermanager / MemcacheStorage.py View on Github external
"""
@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 '{}'