Redis and Python

Redis test memo in Python.

installation:

sudo port install redis
sudo port intall py-redis
mkdir ~/.redis
touch ~/.redis.conf

start redis server:

redis-server

test:

insert:
python redistest.py insert

read:
python redistest.py read
page0
page1
page2
page3
page4
page5
page6
page7
page8
page9

pub:
python redistest.py publish 'hello, hoge'
python redistest.py publish 'bye, page'

sub:
python redistest.py subscribe
{'pattern': None, 'type': 'subscribe', 'channel': '2channel', 'data': 1L}
{'pattern': None, 'type': 'message', 'channel': '2channel', 'data': 'hello, hoge'}
{'pattern': None, 'type': 'message', 'channel': '2channel', 'data': 'bye, page'}

test script:

import os
import sys
import redis

channel='2channel'

def connect():
	# use default port, db index=0
	r = redis.Redis(host='localhost', port=6379, db=0)
	return r


def data_delete():
	r = connect()
	# delete all data in all databases on the current host
	r.flushall()

def data_insert():
	r = connect()
	for i in range(0,10):
		r.set('hoge%d' % i,'page%d' % i)


def data_read():
	r = connect()
	for i in range(0,10):
		print r.get('hoge%d' % i)

def data_publish(message):
	global channel
	r = connect()
	r.publish(channel, message)

def data_subscribe():
	global channel
	r = connect()
	ps = r.pubsub()
	ps.subscribe(channel)
	for message in ps.listen():
		print message



def main():
	functions = {
			"delete" : (lambda x:data_delete()),
			"insert" : (lambda x:data_insert()),
			"read" : (lambda x:data_read()),
			"publish" : (lambda x:data_publish(x)),
			"subscribe" : (lambda x:data_subscribe()),
			}

	param1 = sys.argv[2] if len(sys.argv)>2 else None
	functions[sys.argv[1]](param1)

main()

Leave a Reply

Your email address will not be published. Required fields are marked *