Merge pull request #11 from Overseas-Student-Living/allow-field-based-queries

Allow field or aggregate based queries

Author: juliotrigo

CHANGELOG.rst

@@ -6,6 +6,14 @@ versions, where semantic versioning is used: *major.minor.patch*.
 
 Backwards-compatible changes increment the minor version number only.
 
+
+Version 0.4.0
+-------------
+
+Released 2017-06-21
+
+* Adds support for queries based on model fields or aggregate functions.
+
 Version 0.3.0
 -------------
 

README.rst

@@ -51,6 +51,14 @@ Then we can apply filters to that ``query`` object (multiple times):
 
     result = filtered_query.all()
 
+Note that we can also apply filters to queries defined by fields or functions:
+
+.. code-block:: python
+
+    query_alt_1 = self.session.query(Foo.id, Foo.name)
+    query_alt_2 = self.session.query(func.count(Foo.id))
+
+
 Sort
 ----
 

setup.py

@@ -12,7 +12,7 @@
 
 setup(
     name='sqlalchemy-filters',
-    version='0.3.0',
+    version='0.4.0',
     description='A library to filter SQLAlchemy queries.',
     long_description=readme,
     author='Student.com',

sqlalchemy_filters/models.py

@@ -37,6 +37,6 @@ def get_query_models(query):
         A dictionary with all the models included in the query.
     """
     return {
-        entity['type'].__name__: entity['type']
-        for entity in query.column_descriptions
+        col_desc['entity'].__name__: col_desc['entity']
+        for col_desc in query.column_descriptions
     }

test/interface/test_filters.py

@@ -3,6 +3,7 @@
 import datetime
 
 import pytest
+from sqlalchemy import func
 from sqlalchemy_filters import apply_filters
 from sqlalchemy_filters.exceptions import (
     BadFilterFormat, FieldNotFound, BadQuery
@@ -240,6 +241,32 @@ def test_a_single_dict_can_be_supplied_as_filters(self, session):
         assert result[1].name == 'name_1'
 
 
+class TestApplyFilterOnFieldBasedQuery(TestFiltersMixin):
+
+    @pytest.mark.usefixtures('multiple_bars_inserted')
+    def test_apply_filter_on_single_field_query(self, session):
+        query = session.query(Bar.id)
+        filters = [{'field': 'name', 'op': '==', 'value': 'name_1'}]
+
+        filtered_query = apply_filters(query, filters)
+        result = filtered_query.all()
+
+        assert len(result) == 2
+        assert result[0] == (1,)
+        assert result[1] == (3,)
+
+    @pytest.mark.usefixtures('multiple_bars_inserted')
+    def test_apply_filter_on_aggregate_query(self, session):
+        query = session.query(func.count(Bar.id))
+        filters = [{'field': 'name', 'op': '==', 'value': 'name_1'}]
+
+        filtered_query = apply_filters(query, filters)
+        result = filtered_query.all()
+
+        assert len(result) == 1
+        assert result[0] == (2,)
+
+
 class TestApplyEqualToFilter(TestFiltersMixin):
 
     @pytest.mark.parametrize('operator', ['==', 'eq'])

test/interface/test_models.py

@@ -1,5 +1,6 @@
+from sqlalchemy import func
 from sqlalchemy_filters import get_query_models
-from test.models import Bar, Qux
+from test.models import Bar, Foo, Qux
 
 
 class TestGetQueryModels(object):
@@ -31,3 +32,24 @@ def test_query_with_duplicated_models(self, session):
         entities = get_query_models(query)
 
         assert {'Bar': Bar, 'Qux': Qux} == entities
+
+    def test_query_with_one_field(self, session):
+        query = session.query(Foo.id)
+
+        entities = get_query_models(query)
+
+        assert {'Foo': Foo} == entities
+
+    def test_query_with_multiple_fields(self, session):
+        query = session.query(Foo.id, Bar.id, Bar.name)
+
+        entities = get_query_models(query)
+
+        assert {'Foo': Foo, 'Bar': Bar} == entities
+
+    def test_query_with_aggregate_func(self, session):
+        query = session.query(func.count(Foo.id))
+
+        entities = get_query_models(query)
+
+        assert {'Foo': Foo} == entities