Support Distributed DDL create table queries

docs/models_and_databases.md

@@ -145,6 +145,9 @@ Using the `Database` instance you can create a table for your model, and insert
 
     db.create_table(Person)
     db.insert([dan, suzy])
+
+Including the `cluster` parameter in `create_table` uses a [Distributed DDL](https://clickhouse.tech/docs/en/sql-reference/distributed-ddl/) to create the table on
+ the cluster.
 
 The `insert` method can take any iterable of model instances, but they all must belong to the same model class.
 

setup.py

@@ -0,0 +1,55 @@
+
+SETUP_INFO = dict(
+    name = 'infi.clickhouse_orm',
+    version = '2.1.0.post9',
+    author = 'Jake Lazarus',
+    author_email = 'jake@replicahq.com',
+
+    url = 'https://github.com/Infinidat/infi.clickhouse_orm',
+    license = 'BSD',
+    description = """A Python library for working with the ClickHouse database""",
+
+    # http://pypi.python.org/pypi?%3Aaction=list_classifiers
+    classifiers = [
+        "Intended Audience :: Developers",
+        "Intended Audience :: System Administrators",
+        "License :: OSI Approved :: BSD License",
+        "Operating System :: OS Independent",
+        "Programming Language :: Python",
+        "Programming Language :: Python :: 2.7",
+        "Programming Language :: Python :: 3.4",
+        "Topic :: Software Development :: Libraries :: Python Modules",
+        "Topic :: Database"
+    ],
+
+    install_requires = [
+'iso8601 >= 0.1.12',
+'pytz',
+'requests',
+'setuptools'
+],
+    namespace_packages = ['infi'],
+
+    package_dir = {'': 'src'},
+    package_data = {'': []},
+    include_package_data = True,
+    zip_safe = False,
+
+    entry_points = dict(
+        console_scripts = [],
+        gui_scripts = [],
+        ),
+)
+
+if SETUP_INFO['url'] is None:
+    _ = SETUP_INFO.pop('url')
+
+def setup():
+    from setuptools import setup as _setup
+    from setuptools import find_packages
+    SETUP_INFO['packages'] = find_packages('src')
+    _setup(**SETUP_INFO)
+
+if __name__ == '__main__':
+    setup()
+

src/infi/clickhouse_orm/database.py

@@ -138,15 +138,15 @@ def drop_database(self):
         self._send('DROP DATABASE `%s`' % self.db_name)
         self.db_exists = False
 
-    def create_table(self, model_class):
+    def create_table(self, model_class, cluster=None):
         '''
         Creates a table for the given model class, if it does not exist already.
         '''
         if model_class.is_system_model():
             raise DatabaseException("You can't create system table")
         if getattr(model_class, 'engine') is None:
             raise DatabaseException("%s class must define an engine" % model_class.__name__)
-        self._send(model_class.create_table_sql(self))
+        self._send(model_class.create_table_sql(self, cluster=cluster))
 
     def drop_table(self, model_class):
         '''

src/infi/clickhouse_orm/models.py

@@ -348,11 +348,12 @@ def has_funcs_as_defaults(cls):
         return cls._has_funcs_as_defaults
 
     @classmethod
-    def create_table_sql(cls, db):
+    def create_table_sql(cls, db, cluster):
         '''
         Returns the SQL statement for creating a table for this model.
         '''
-        parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` (' % (db.db_name, cls.table_name())]
+        parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` %s (' % (
+            db.db_name, cls.table_name(), 'ON CLUSTER `%s`' % (cluster) if cluster else '')]
         # Fields
         items = []
         for name, field in cls.fields().items():
@@ -483,12 +484,13 @@ def is_system_model(cls):
 class BufferModel(Model):
 
     @classmethod
-    def create_table_sql(cls, db):
+    def create_table_sql(cls, db, cluster):
         '''
         Returns the SQL statement for creating a table for this model.
         '''
-        parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` AS `%s`.`%s`' % (db.db_name, cls.table_name(), db.db_name,
-                                                                        cls.engine.main_model.table_name())]
+        parts = ['CREATE TABLE IF NOT EXISTS `{0}`.`{1}` {2} AS `{0}`.`{3}`'.format(
+            db.db_name, cls.table_name(), 'ON CLUSTER `%s`' % (cluster) if cluster else '',
+            cls.engine.main_model.table_name())]
         engine_str = cls.engine.create_table_sql(db)
         parts.append(engine_str)
         return ' '.join(parts)
@@ -506,12 +508,13 @@ class MergeModel(Model):
     _table = StringField(readonly=True)
 
     @classmethod
-    def create_table_sql(cls, db):
+    def create_table_sql(cls, db, cluster):
         '''
         Returns the SQL statement for creating a table for this model.
         '''
         assert isinstance(cls.engine, Merge), "engine must be an instance of engines.Merge"
-        parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` (' % (db.db_name, cls.table_name())]
+        parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` %s (' % (
+            db.db_name, cls.table_name(), 'ON CLUSTER `%s`' % (cluster) if cluster else '')]
         cols = []
         for name, field in cls.fields().items():
             if name != '_table':
@@ -590,7 +593,7 @@ def fix_engine_table(cls):
         cls.engine.table = storage_models[0]
 
     @classmethod
-    def create_table_sql(cls, db):
+    def create_table_sql(cls, db, cluster):
         '''
         Returns the SQL statement for creating a table for this model.
         '''
@@ -599,8 +602,8 @@ def create_table_sql(cls, db):
         cls.fix_engine_table()
 
         parts = [
-            'CREATE TABLE IF NOT EXISTS `{0}`.`{1}` AS `{0}`.`{2}`'.format(
-                db.db_name, cls.table_name(), cls.engine.table_name),
+            'CREATE TABLE IF NOT EXISTS `{0}`.`{1}` {3} AS `{0}`.`{2}`'.format(
+                db.db_name, cls.table_name(), cls.engine.table_name, 'ON CLUSTER `%s`' % (cluster) if cluster else ''),
             'ENGINE = ' + cls.engine.create_table_sql(db)]
         return '\n'.join(parts)