From 843462005c7b7bd3a009d2edb8db513fec3c7c17 Mon Sep 17 00:00:00 2001 From: ducdetronquito Date: Mon, 20 Feb 2017 19:38:17 +0100 Subject: [PATCH] [#31] Can order by a SelectQuery. --- plume/plume.py | 41 +++++++++++---- tests/test_selectquery.py | 104 +++++++++++++++++++++++++++++++++++++- 2 files changed, 134 insertions(+), 11 deletions(-) diff --git a/plume/plume.py b/plume/plume.py index 0514796..165b422 100644 --- a/plume/plume.py +++ b/plume/plume.py @@ -71,10 +71,13 @@ class SQLiteDB: # Select Query ALL = '*' + ASC = 'ASC' + DESC = 'DESC' DISTINCT = 'DISTINCT' FROM = 'FROM' LIMIT = 'LIMIT' OFFSET = 'OFFSET' + ORDER_BY = 'ORDER BY' SELECT = 'SELECT' WHERE = 'WHERE' @@ -171,6 +174,9 @@ def build_select(self, query): if query._offset is not None: output.extend((self.OFFSET, str(query._offset))) + if query._order_by: + output.extend((self.ORDER_BY, str(CSV(query._order_by)))) + return ' '.join(output) def build_update(self, query): @@ -248,13 +254,13 @@ def __new__(cls, clsname, bases, attrs): attrs['__slots__'] = ('_values',) # Create the new class. - new_class = super().__new__(cls, clsname, bases, attrs) + model = super().__new__(cls, clsname, bases, attrs) # Each field of the class knows its related model class name. - for fieldname in new_class._fieldnames: - getattr(new_class, fieldname).model_name = clsname.lower() + for fieldname in model._fieldnames: + getattr(model, fieldname).model = model - return new_class + return model class Node: __slots__ = () @@ -312,13 +318,13 @@ def __str__(self): class Field(Node): - __slots__ = ('default', 'model_name', 'name', 'required', 'unique', 'value') + __slots__ = ('default', 'model', 'name', 'required', 'unique', 'value') internal_type = None sqlite_datatype = None def __init__(self, default=None, name=None, required=True, unique=False): self.default = default - self.model_name = None + self.model = None self.name = name self.required = required self.unique = unique @@ -354,7 +360,13 @@ def __set__(self, instance, value): instance._values._replace(**{self.name: value}) def __str__(self): - return '.'.join((self.model_name, self.name)) + return '.'.join((self.model.__name__.lower(), self.name)) + + def asc(self): + return ' '.join((str(self), self.model._db.ASC)) + + def desc(self): + return ' '.join((str(self), self.model._db.DESC)) def is_valid(self, value): """Return True if the provided value match the internal field.""" @@ -651,16 +663,21 @@ class SelectQuery(FilterableQuery): clause. The user is allowed to add dynamically several criteria on a QuerySet. The SelectQuery only hit the database when it is iterated over or sliced. """ - __slots__ = ('_db', '_distinct', '_fields', '_limit', '_model', '_offset', '_tables') + __slots__ = ( + '_db', '_distinct', '_fields', '_limit', '_model', + '_offset', '_order_by', '_tables' + ) def __init__(self, db): super().__init__() self._db = db - self._tables = [] + self._distinct = False self._fields = [] self._limit = None self._offset = None - self._distinct = False + self._order_by = [] + self._tables = [] + def __str__(self): return ''.join(('(', self.build(), ')')) @@ -765,6 +782,10 @@ def offset(self, offset:int): self._offset = offset return self + def order_by(self, *fields): + self._order_by.extend(field if isinstance(field, str) else str(field) for field in fields) + return self + def select(self, *fields): # Allow to filter Select-Query on columns. self._fields.extend(field if isinstance(field, str) else str(field) for field in fields) diff --git a/tests/test_selectquery.py b/tests/test_selectquery.py index f8eec0b..127c881 100644 --- a/tests/test_selectquery.py +++ b/tests/test_selectquery.py @@ -11,7 +11,10 @@ def test_is_slotted(self): SelectQuery(self.db).__dict__ def test_attributes(self): - expected = ('_db', '_distinct', '_fields', '_limit', '_model', '_offset', '_tables') + expected = ( + '_db', '_distinct', '_fields', '_limit', '_model', + '_offset', '_order_by', '_tables' + ) result = SelectQuery(self.db).__slots__ assert result == expected @@ -35,6 +38,9 @@ def test_has_limit_method(self): def test_has_offset_method(self): assert hasattr(SelectQuery(self.db), 'offset') is True + def test_has_order_by_method(self): + assert hasattr(SelectQuery(self.db), 'order_by') is True + def test_can_be_iterated(self): assert hasattr(SelectQuery(self.db), '__iter__') is True @@ -339,3 +345,99 @@ def test_can_select_with_exists_return_true(self): expected = 1 assert result[0][0] == expected + +class TestSelectQueryOrderBy(BaseTestCase): + + def test_can_order_by_one_field(self): + query = SelectQuery(self.db).tables(Trainer).order_by(Trainer.name).build() + expected = 'SELECT * FROM trainer ORDER BY trainer.name' + assert query == expected + + def test_result_order_by_one_field(self): + self.add_trainer(['Giovanni', 'Jessie']) + result = ( + SelectQuery(self.db).select(Trainer.name).tables(Trainer) + .order_by(Trainer.name).execute() + ) + assert result[0][0] == 'Giovanni' + assert result[1][0] == 'Jessie' + + def test_can_order_by_one_field_as_string(self): + query = SelectQuery(self.db).tables(Trainer).order_by('name').build() + expected = 'SELECT * FROM trainer ORDER BY name' + assert query == expected + + def test_result_order_by_one_field_as_string(self): + self.add_trainer(['Giovanni', 'Jessie']) + result = ( + SelectQuery(self.db).select(Trainer.name).tables(Trainer) + .order_by('name').execute() + ) + assert result[0][0] == 'Giovanni' + assert result[1][0] == 'Jessie' + + def test_can_order_by_one_field_with_sort_order_asc(self): + query = SelectQuery(self.db).tables(Trainer).order_by(Trainer.name.asc()).build() + expected = 'SELECT * FROM trainer ORDER BY trainer.name ASC' + assert query == expected + + def test_result_order_by_one_field_with_sort_order_asc(self): + self.add_trainer(['Giovanni', 'Jessie']) + result = ( + SelectQuery(self.db).select(Trainer.name).tables(Trainer) + .order_by(Trainer.name).execute() + ) + assert result[0][0] == 'Giovanni' + assert result[1][0] == 'Jessie' + + def test_can_order_by_one_field_with_sort_order_desc(self): + query = SelectQuery(self.db).tables(Trainer).order_by(Trainer.name.desc()).build() + expected = 'SELECT * FROM trainer ORDER BY trainer.name DESC' + assert query == expected + + def test_result_order_by_one_field_with_sort_order_desc(self): + self.add_trainer(['Giovanni', 'Jessie']) + result = ( + SelectQuery(self.db).select(Trainer.name).tables(Trainer) + .order_by(Trainer.name.desc()).execute() + ) + assert result[0][0] == 'Jessie' + assert result[1][0] == 'Giovanni' + + def test_can_order_by_several_fields(self): + query = SelectQuery(self.db).tables(Trainer).order_by(Trainer.name, Trainer.age).build() + expected = 'SELECT * FROM trainer ORDER BY trainer.name, trainer.age' + assert query == expected + + def test_result_order_by_several_fields(self): + InsertQuery(self.db).table(Trainer).from_dicts([ + {'name': 'Jessie', 'age': 17}, + {'name': 'Giovanni', 'age': 66}, + {'name': 'Giovanni', 'age': 42} + ]).execute() + + result = ( + SelectQuery(self.db).select(Trainer.name, Trainer.age).tables(Trainer) + .order_by(Trainer.name, Trainer.age).execute() + ) + assert result == [('Giovanni', 42), ('Giovanni', 66), ('Jessie', 17)] + + def test_can_order_by_several_fields_with_sort_order(self): + query = SelectQuery(self.db).tables(Trainer).order_by( + Trainer.name.asc(), Trainer.age.desc() + ).build() + expected = 'SELECT * FROM trainer ORDER BY trainer.name ASC, trainer.age DESC' + assert query == expected + + def test_result_order_by_several_fields_with_sort_order(self): + InsertQuery(self.db).table(Trainer).from_dicts([ + {'name': 'Jessie', 'age': 17}, + {'name': 'Giovanni', 'age': 66}, + {'name': 'Giovanni', 'age': 42} + ]).execute() + + result = ( + SelectQuery(self.db).select(Trainer.name, Trainer.age).tables(Trainer) + .order_by(Trainer.name.asc(), Trainer.age.desc()).execute() + ) + assert result == [('Giovanni', 66), ('Giovanni', 42), ('Jessie', 17)]