python 分享一个用SSDB 做缓存的类
🕣 by pyList at 2015-07-02 19:59
分享一个用SSDB 做缓存的类,使用python sdk,可以作为一个服务跑起来。
__doc__ = """
ssdb_cache has a dictionary like interface and a SSDB(http://ssdb.io/) backend
It uses pickle to store Python objects and strings
Multithreading is supported
"""
import datetime
# pip install pyssdb --upgrade
import pyssdb
try:
import cPickle as pickle
except ImportError:
import pickle
class CACHE:
"""Stores and retrieves persistent data through a dict-like interface
Data is stored on disk using SSDB
cache_name/filename:
The name of Hashmap, use this name "filename" to be compatible with PersistentDict in webscraping library
expires:
A timedelta object of how old data can be before expires. By default is set to None to disable.
host:
The SSDB server host
port:
The SSDB server port
"""
def __init__(self, cache_name=None, filename=None, expires=None, host='127.0.0.1', port=8888):
"""initialize a new CACHE
"""
self.cache_name = filename or cache_name or 'cache'
self.expires = expires
self.ssdb_host = host
self.ssdb_port = port
self.ssdb_client = pyssdb.Client(host=self.ssdb_host, port=self.ssdb_port)
def __copy__(self):
"""make a copy of current cache settings
"""
return CACHE(cache_name=self.cache_name, expires=self.expires, host=self.ssdb_host, port=self.ssdb_port)
def __contains__(self, key):
"""check the database to see if a key exists
"""
return self.ssdb_client.hexists(self.cache_name, key)
def __getitem__(self, key):
"""return the value of the specified key or raise KeyError if not found
"""
value_meta_update = self.ssdb_client.hget(self.cache_name, key)
if value_meta_update is not None:
value_meta_update = self.deserialize(value_meta_update)
if self.is_fresh(value_meta_update.get('updated')):
return value_meta_update.get('value')
else:
raise KeyError("Key `%s' is stale" % key)
else:
raise KeyError("Key `%s' does not exist" % key)
def __delitem__(self, key):
"""remove the specifed value from the database
"""
self.ssdb_client.hdel(self.cache_name, key)
def __setitem__(self, key, value):
"""set the value of the specified key
"""
value_meta_update = {'value': value, 'meta': None, 'updated': datetime.datetime.now()}
self.ssdb_client.hset(self.cache_name, key, self.serialize(value_meta_update))
def __len__(self):
"""get the number of elements in CACHE
"""
return self.ssdb_client.hsize(self.cache_name)
def serialize(self, value):
"""convert object to a pickled string to save in the db
"""
return pickle.dumps(value, protocol=pickle.HIGHEST_PROTOCOL)
def deserialize(self, value):
"""convert pickled string from database back into an object
"""
return pickle.loads(value)
def get(self, key, default=None):
"""Get data at key and return default if not defined
"""
data = default
if key:
value_meta_update = self.ssdb_client.hget(self.cache_name, key)
if value_meta_update is not None:
return self.deserialize(value_meta_update)
return data
def meta(self, key, value=None):
"""Get / set meta for this value
if value is passed then set the meta attribute for this key
if not then get the existing meta data for this key
"""
if value is None:
# want to get meta
value_meta_update = self.ssdb_client.hget(self.cache_name, key)
if value_meta_update is not None:
return self.deserialize(value_meta_update).get('meta')
else:
raise KeyError("Key `%s' does not exist" % key)
else:
# want to set meta
value_meta_update = self.ssdb_client.hget(self.cache_name, key)
if value_meta_update is not None:
value_meta_update = self.deserialize(value_meta_update)
value_meta_update['meta'] = value
value_meta_update['updated'] = datetime.datetime.now()
self.ssdb_client.hset(self.cache_name, key, self.serialize(value_meta_update))
def clear(self):
"""Clear all cached data in this collecion
"""
self.ssdb_client.hclear(self.cache_name)
def is_fresh(self, t):
"""returns whether this datetime has expired
"""
return self.expires is None or datetime.datetime.now() - t < self.expires
if __name__ == '__main__':
cache = CACHE(cache_name='test')
print 'Name' in cache
cache['Name'] = 'Peng Qi'
print cache['Name']
print cache.get('Name')
cache.meta('Name', {'Age': 29})
print cache.meta('Name')
本文网址: https://pylist.com/t/1435838386 (转载注明出处)
如果你有任何建议或疑问可以在下面 留言
发表第一条评论!
相关推荐
小工具
标签
python
ssdb
接口
提速
时间
判断
当前
是否
文件类型
magic
从文件
命令行
参数
处理
sqlite
数据库
经验
示例
sae
tornado
pytenjin
朋友圈
微信
好友
识别
分享
练手
简单
准备
项目
一个
转码
链接
中文
安全
linux
engine
app
google
openwrt
国内
form
data
post
上传
u盘
mac
diskutil
格式化
顽固
最新版
wndr4300
固件
升级
sqlite3
连接池
time
golang
timestamp
小结
默认值
struct
设置
failed
load
devtools
sourcemap
chrome
cpu
debian
ubuntu
查看
温度
全文
搜索
实现
身份验证
authenticator
迁移
手机
硬件加速
ubnt
er
路由
编译
cjson
lua
module
错误
解决
自动更新
microsoft
关闭
store
登录
未知
静音
风扇
主机
笔记本
改造
低功耗
web
爬虫
服务器
组装
尝鲜
视频
体验
server
浏览
webdriver
微博
selenium
gnu
安装
系统启动
usb
宅家
坑记
屏幕
动手
nginx
quic
抢先
最近发表
- Chrome 控制台 DevTools failed to load SourceMap 警告的消除方法
- Mac 关闭 Microsoft 自动更新
- Mac 登录 App Store 出现“发生了未知错误”的解决方法
- 老笔记本改造为无风扇静音主机方案
- 自己组装21瓦低功耗家庭爬虫、文件、web服务器
- 微信视频号尝鲜体验
- Ubuntu/Debian 查看CPU温度的方法
- 在Ubuntu/debian Server 系统使用Chrome 无头浏览模式
- 换手机后 Google 身份验证器 Google Authenticator 数据迁移的简单方法
- 使用Golang selenium WebDriver 自动登录微博
- 在 Ubuntu 或其它 GNU/Linux 系统下安装 Debian
- Mac 下制作 USB ubuntu/debian 系统启动、安装盘的几种方法
- ubuntu/debian 下自行编译 OpenWRT 固件
- 宅家自己动手换手机屏幕掉坑记
- 路由 UBNT ER-X 官方固件升级及开启硬件加速的方法
- 在 Nginx 和 Golang web 上抢先体验 QUIC
最近浏览
- OpenWrt 国内源
- python form-data post上传数据简便方法
- 在终端使用Mac diskutil 命令格式化顽固U盘
- WNDR4300 固件升级到 OpenWrt 最新版
- python SQLite3 连接池
- golang timestamp time 时间戳小结
- go struct 设置默认值
- Chrome 控制台 DevTools failed to load SourceMap 警告的消除方法
- Ubuntu/Debian 查看CPU温度的方法
- ssdb 全文搜索的实现
- 换手机后 Google 身份验证器 Google Authenticator 数据迁移的简单方法
- 路由 UBNT ER-X 官方固件升级及开启硬件加速的方法
- 用python 对中文链接安全转码
- ubuntu/debian 下自行编译 OpenWRT 固件
- 出现 lua module 'cjson' not found 错误的解决方法
- Mac 关闭 Microsoft 自动更新