Merge branch 'develop' of https://github.com/emakarov/infi.clickhouse_orm into emakarov-develop

# Conflicts:
#	src/infi/clickhouse_orm/models.py
#	tests/test_database.py
This commit is contained in:
Itai Shirav 2017-02-19 09:12:00 +02:00
commit 1d573ded96
4 changed files with 85 additions and 1 deletions

View File

@ -380,6 +380,30 @@ For a ``SummingMergeTree`` you can optionally specify the summing columns::
engine = engines.SummingMergeTree('EventDate', ('OrderID', 'EventDate', 'BannerID'),
summing_cols=('Shows', 'Clicks', 'Cost'))
A ``Buffer`` engine is available for BufferModels. (See below how to use BufferModel). You can specify following parameters::
engine = engines.Buffer(Person) # you need to initialize engine with main Model. Other default parameters will be used
# or:
engine = engines.Buffer(Person, num_layers=16, min_time=10,
max_time=100, min_rows=10000, max_rows=1000000,
min_bytes=10000000, max_bytes=100000000)
Buffer Models
-------------
Here's how you can define Model for Buffer Engine. The Buffer Model should be inherited from models.BufferModel and main Model::
class PersonBuffer(models.BufferModel, Person):
engine = engines.Buffer(Person)
Then you can insert objects into Buffer model and they will be handled by Clickhouse properly::
db.create_table(PersonBuffer)
suzy = PersonBuffer(first_name='Suzy', last_name='Jones')
dan = PersonBuffer(first_name='Dan', last_name='Schwartz')
db.insert([dan, suzy])
Data Replication
****************
@ -401,6 +425,7 @@ After cloning the project, run the following commands::
To run the tests, ensure that the ClickHouse server is running on http://localhost:8123/ (this is the default), and run::
bin/nosetests
=======
To see test coverage information run::

View File

@ -62,3 +62,30 @@ class SummingMergeTree(MergeTree):
params.append('(%s)' % ', '.join(self.summing_cols))
return params
class Buffer(Engine):
"""Here we define Buffer engine
Read more here https://clickhouse.yandex/reference_en.html#Buffer
"""
#Buffer(database, table, num_layers, min_time, max_time, min_rows, max_rows, min_bytes, max_bytes)
def __init__(self, main_model, num_layers=16, min_time=10, max_time=100, min_rows=10000, max_rows=1000000, min_bytes=10000000, max_bytes=100000000):
self.main_model = main_model
self.num_layers = num_layers
self.min_time = min_time
self.max_time = max_time
self.min_rows = min_rows
self.max_rows = max_rows
self.min_bytes = min_bytes
self.max_bytes = max_bytes
def create_table_sql(self, db_name):
# Overriden create_table_sql example:
#sql = 'ENGINE = Buffer(merge, hits, 16, 10, 100, 10000, 1000000, 10000000, 100000000)'
sql = 'ENGINE = Buffer(`%s`, `%s`, %d, %d, %d, %d, %d, %d, %d)' % (
db_name, self.main_model.table_name(), self.num_layers,
self.min_time, self.max_time, self.min_rows,
self.max_rows, self.min_bytes, self.max_bytes
)
return sql

View File

@ -202,3 +202,16 @@ class Model(with_metaclass(ModelBase)):
data = self.__dict__
return {name: data[name] for name, field in fields}
class BufferModel(Model):
@classmethod
def create_table_sql(cls, db_name):
'''
Returns the SQL command for creating a table for this model.
'''
parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` AS `%s`.`%s`' % (db_name, cls.table_name(), db_name, cls.engine.main_model.table_name())]
engine_str = cls.engine.create_table_sql(db_name)
parts.append(engine_str)
return ' '.join(parts)

View File

@ -3,7 +3,7 @@
import unittest
from infi.clickhouse_orm.database import Database, DatabaseException
from infi.clickhouse_orm.models import Model
from infi.clickhouse_orm.models import Model, BufferModel
from infi.clickhouse_orm.fields import *
from infi.clickhouse_orm.engines import *
@ -16,8 +16,10 @@ class DatabaseTestCase(unittest.TestCase):
def setUp(self):
self.database = Database('test-db')
self.database.create_table(Person)
self.database.create_table(PersonBuffer)
def tearDown(self):
self.database.drop_table(PersonBuffer)
self.database.drop_table(Person)
self.database.drop_database()
@ -27,6 +29,10 @@ class DatabaseTestCase(unittest.TestCase):
for instance in data:
self.assertEquals(self.database, instance.get_database())
def _insert_and_check_buffer(self, data, count):
self.database.insert(data)
self.assertEquals(count, self.database.count(PersonBuffer))
def test_insert__generator(self):
self._insert_and_check(self._sample_data(), len(data))
@ -137,6 +143,9 @@ class DatabaseTestCase(unittest.TestCase):
self.database.drop_database()
self.database = orig_database
def test_insert_buffer(self):
self._insert_and_check_buffer(self._sample_buffer_data(), len(data))
def _sample_data(self):
for entry in data:
yield Person(**entry)
@ -160,6 +169,11 @@ class DatabaseTestCase(unittest.TestCase):
with self.assertRaises(DatabaseException):
self.database.drop_table(ReadOnlyModel)
def _sample_buffer_data(self):
for entry in data:
yield PersonBuffer(**entry)
class Person(Model):
@ -177,6 +191,11 @@ class ReadOnlyModel(Model):
name = StringField()
class PersonBuffer(BufferModel, Person):
engine = Buffer(Person)
data = [
{"first_name": "Abdul", "last_name": "Hester", "birthday": "1970-12-02", "height": "1.63"},