操作系统:ubuntu 12.04 TLS
python版本:2.7.5
准备工作:
1、安装tornado
执行如下命令:
[code]
sudo easy_install tornado
[/code]
2、安装redis
执行如下命令:
[code]
cd /tmp
wget http://redis.googlecode.com/files/redis-2.2.13.tar.gz
tar -zxf redis-2.2.13.tar.gz
cd redis-2.2.13
make
sudo make install
[/code]
启动redis的命令:
[code]
sudo /etc/init.d/redis-server start
[/code]
关闭redis的命令:
[code]
sudo /etc/init.d/redis-server stop
[/code]
重启redis的命令:
[code]
sudo /etc/init.d/redis-server restart
[/code]
3、安装python连接redis的模块
点此下载安装源码
下载源码解压,运行如下命令:
[code]
sudo python setup.py install
[/code]
4、安装ujson
运行如下命令:
[code]
sudo easy_install ujson
[/code]
好了,写一个session文件,基于redis数据库:
session.py
[code lang=”python”]
#/usr/bin/python
# coding: utf-8
import uuid
import hmac
import ujson
import hashlib
import redis
__author__ = ‘karldoenitz’
"""
需要安装的第三方库:
python的redis连接:http://pypi.python.org/pypi?%3Aaction=search&term=redis&submit=search
需要ujson
安装方法:sudo easy_install ujson
"""
class SessionData(dict):
"""
hmac:
Hash-based message authentication code,利用哈希算法,以一个密钥和一个消息为输入,生成一个消息摘要作为输出
"""
def __init__(self, session_id, hmac_key):
self.session_id = session_id
self.hmac_key = hmac_key
class Session(SessionData):
def __init__(self, session_manager, request_handler):
self.session_manager = session_manager
self.request_handler = request_handler
try:
current_session = session_manager.get(request_handler)
except InvalidSessionException:
current_session = session_manager.get()
for key, data in current_session.iteritems():
self[key] = data
self.session_id = current_session.session_id
self.hmac_key = current_session.hmac_key
def save(self):
self.session_manager.set(self.request_handler, self)
class SessionManager(object):
"""
redis的地址,端口和密码
"""
def __init__(self, secret, store_options, session_timeout):
self.secret = secret
self.session_timeout = session_timeout
try:
if store_options[‘redis_pass’]:
self.redis = redis.StrictRedis(host=store_options[‘redis_host’], port=store_options[‘redis_port’], password=store_options[‘redis_pass’])
else:
self.redis = redis.StrictRedis(host=store_options[‘redis_host’], port=store_options[‘redis_port’])
except Exception as e:
print e
def _fetch(self, session_id):
try:
session_data = raw_data = self.redis.get(session_id)
if raw_data:
self.redis.setex(session_id, self.session_timeout, raw_data)
session_data = ujson.loads(raw_data)
if isinstance(session_data, dict):
return session_data
else:
return {}
except IOError:
return {}
def get(self, request_handler=None):
if not request_handler:
session_id = None
hmac_key = None
else:
session_id = request_handler.get_secure_cookie("session_id")
hmac_key = request_handler.get_secure_cookie("verification")
if not session_id:
session_exists = False
session_id = self._generate_id()
hmac_key = self._generate_hmac(session_id)
else:
session_exists = True
check_hmac = self._generate_hmac(session_id)
if hmac_key != check_hmac:
raise InvalidSessionException()
session = SessionData(session_id, hmac_key)
if session_exists:
session_data = self._fetch(session_id)
for key, data in session_data.iteritems():
session[key] = data
return session
def set(self, request_handler, session):
request_handler.set_secure_cookie("session_id", session.session_id)
request_handler.set_secure_cookie("verification", session.hmac_key)
session_data = ujson.dumps(dict(session.items()))
self.redis.setex(session.session_id, self.session_timeout, session_data)
def _generate_id(self):
new_id = hashlib.sha256(self.secret + str(uuid.uuid4()))
return new_id.hexdigest()
def _generate_hmac(self, session_id):
return hmac.new(session_id, self.secret, hashlib.sha256).hexdigest()
#此处可以自定义异常处理
class InvalidSessionException(Exception):
pass
[/code]
接下来我们写一个tornado项目来测试一下我们的session,文件结构如下:
├── hello_world.py
├── base.py
├── session.py
├── templates
│ └── hello.html
└── testhandler.py
base.py
[code lang=”python”]
#coding=utf-8
__author__ = ‘karldoenitz’
import tornado.web
import session
class BaseHandler(tornado.web.RequestHandler):
def __init__(self, *argc, **argkw):
super(BaseHandler, self).__init__(*argc, **argkw)
self.session = session.Session(self.application.session_manager, self)
[/code]
testhandler.py
[code lang=”python”]
#coding=utf-8
__author__ = ‘karldoenitz’
from base import BaseHandler
class HelloHandler(BaseHandler):
def get(self):
l = []
hello = self.get_argument(‘non1’)
world = self.get_argument(‘non2’)
print self.session
l.append(hello)
l.append(world)
l.append(self.session[‘first_session_value’])
self.render("./templates/hello.html", page_object=l)
class TestGetHandler(BaseHandler):
def get(self):
test = self.get_argument(‘test’, ”)
self.session[‘first_session_value’] = "今天玩session玩到晚上11点多"
self.session.save()
print self.session
self.write(test)
self.finish()
[/code]
hello_world.py
[code lang=”python”]
#coding=utf-8
__author__ = ‘karldoenitz’
import os
import sys
from session import *
from tornado.ioloop import IOLoop
import tornado
import session
from testhandler import HelloHandler, TestGetHandler
from tornado.web import Application, url
from tornado.options import define, options
reload(sys)
sys.setdefaultencoding(‘utf-8’)
from base import BaseHandler
from tornado.web import HTTPError
def login_required(f):
def _wrapper(self, *args, **kwargs):
print self.get_current_user()
logged = self.get_current_user()
if logged == None:
self.write(‘no login’)
self.finish()
else:
ret = f(self, *args, **kwargs)
return _wrapper
class Application(tornado.web.Application):
def __init__(self):
settings = dict(
cookie_secret="e446976943b4e8442f099fed1f3fea28462d5832f483a0ed9a3d5d3859f==78d",
session_secret="3cdcb1f00803b6e78ab50b466a40b9977db396840c28307f428b25e2277f1bcc",
session_timeout=60,
store_options={
‘redis_host’: ‘localhost’,
‘redis_port’: 6379,
‘redis_pass’: ”
},
)
handlers = [
(r"/test", TestGetHandler),
(r"/hello", HelloHandler)
]
tornado.web.Application.__init__(self, handlers, **settings)
self.session_manager = session.SessionManager(settings["session_secret"], settings["store_options"], settings["session_timeout"])
define(‘port’, default=8888, group=’application’)
if __name__ == "__main__":
application = Application()
print "Application starts on port: ", options.port
application.listen(options.port)
tornado.ioloop.IOLoop.instance().start()
[/code]
运行如下命令:
[code]
python hello_world.py
[/code]
然后访问localhost:8888/test?test=list,页面如下:
再访问http://localhost:8888/hello?non1=1&non2=2,页面如下:
呵呵OK了,记住手一定要快啊哦,我把session的有效期设为60秒,可以改的。