Тест. Сравнение производительности SQLLite и Redis

test.py

#-*-encoding:utf8-*-

from sqlalchemy.orm import *
from sqlalchemy import Table, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from redis import Redis
from stopwatch import *

""" Либа. Запуск тестов. """

MAX = 10 ** 3 # Число записей в тесте

testing = list() # Список тестируемых ф-й
	
def testing_function(f) :
	""" Декоратор. Добавляет ф-ю в список тестируемых. """

	testing.append(f)
	return f
	
# SQLAlchemy тест

# Соединение с SQLite

engine = create_engine('sqlite:///file.db', echo=False)
Session = sessionmaker(bind=engine)
Base = declarative_base()
session = Session()

class SimpleTable(Base) :
	""" Простая таблица для теста. """

	__tablename__ = "simple_table"
	
	id = Column(Integer, primary_key = True)
	value1 = Column(String)
	value2 = Column(String)
	
Base.metadata.create_all(engine) # Создаем БД

# Функции тестирования SQL

@testing_function
def sql_read_test() :
	""" Тест SQLAlchemy на чтение. """
	
	for i in xrange(1, MAX) :
		
		t = session.query(SimpleTable).filter_by(id = i).first()
		v = t.value1
		v = t.value2

@testing_function
def sql_write_test() :
	""" Тест SQLAlchemy на запись. Много записей - один коммит. """

	for i in xrange(1, MAX) :
		# Заполняем таблицу MAX значениями
	
		t = SimpleTable()
		t.value1 = "goodvalue" + str(i)
		t.value2 = "goodvalue" + str(i)
		session.add(t)
	
	session.commit()

@testing_function
def sql_read_query_test() :
	""" Тест SQLAlchemy на запросы. Много записей - много коммитов. """

	for i in xrange(1, MAX) :
		# Заполняем таблицу MAX значениями
	
		t = SimpleTable()
		t.value1 = "goodvalue" + str(i)
		t.value2 = "goodvalue" + str(i)
		session.add(t)
		session.commit()

@testing_function		
def sql_write_query_test() :
	""" Тест SQLAlchemy на запись. Много записей - много коммитов. """

	for i in xrange(1, MAX) :
		# Заполняем таблицу MAX значениями
	
		t = SimpleTable()
		t.value1 = "goodvalue" + str(i)
		t.value2 = "goodvalue" + str(i)
		session.add(t)
		session.commit()
		

# Redis

r_server = Redis(host='localhost', port=6379, db=0)

@testing_function
def redis_read_test() :
	""" Тест Redis на чтение. """ 
	
	for i in xrange(1, MAX) :
		# Заполняем Redis MAX значениями
	
		a = r_server.get("test:%d:key" % i)
		assert a == "value"

@testing_function
def redis_write_test() :
	""" Тест Redis на запись. """

	for i in xrange(1, MAX) :
		# Заполняем Redis MAX значениями
	
		r_server.set("test:%d:key" % i, "value")
		
		

		
if __name__ == "__main__" :
	
	for i, f in enumerate(testing) :
		# Тестируем все функции из списка тестируемых
		clockit(f)()
		print "Test %s complete." % i