Infinidat/infi.clickhouse_orm
1
1
Fork 0
mirror of https://github.com/Infinidat/infi.clickhouse_orm.git synced 2025-08-02 19:20:14 +03:00
Code Issues Packages Projects Releases Wiki Activity

added validation for final and test

Browse Source
This commit is contained in:
solebox 2024-06-03 15:23:11 +03:00
parent 5f2825b9f2
commit c1f14b4dad
2 changed files with 8 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/infi/clickhouse_orm/query.py
View File

@ -538,10 +538,10 @@ class QuerySet(object):
def final(self): def final(self):
""" """
Adds a FINAL modifier to table, meaning data will be collapsed to final version. Adds a FINAL modifier to table, meaning data will be collapsed to final version.
Can be used with the `CollapsingMergeTree` and `ReplacingMergeTree` engines only. Can be used with the `CollapsingMergeTree`, `ReplacingMergeTree` and `AggregatingMergeTree` engines only.
""" """
from .engines import CollapsingMergeTree, ReplacingMergeTree from .engines import CollapsingMergeTree, ReplacingMergeTree, AggregatingMergeTree
if not isinstance(self._model_cls.engine, (CollapsingMergeTree, ReplacingMergeTree)): if not isinstance(self._model_cls.engine, (CollapsingMergeTree, ReplacingMergeTree, AggregatingMergeTree)):
raise TypeError('final() method can be used only with the CollapsingMergeTree and ReplacingMergeTree engines') raise TypeError('final() method can be used only with the CollapsingMergeTree and ReplacingMergeTree engines')
qs = copy(self) qs = copy(self)

5
tests/test_engines.py
View File

@ -73,6 +73,11 @@ class EnginesTestCase(_EnginesHelperTestCase):
engine = ReplacingMergeTree('date', ('date', 'event_id', 'event_group'), 'event_uversion') engine = ReplacingMergeTree('date', ('date', 'event_id', 'event_group'), 'event_uversion')
self._create_and_insert(TestModel) self._create_and_insert(TestModel)
def test_aggregating_merge_tree(self):
class TestModel(SampleModel):
engine = AggregatingMergeTree('date', ('date', 'event_group'))
self._create_and_insert(TestModel)
def test_tiny_log(self): def test_tiny_log(self):
class TestModel(SampleModel): class TestModel(SampleModel):
engine = TinyLog() engine = TinyLog()