Add exclude filter

Author: frmdstryr

CHANGELOG.md

@@ -1,6 +1,7 @@
 # 0.7.5
 
 - Fix a bug preventing select_related on multiple fields from working
+- Add django style `exclude` to filter
 
 # 0.7.4
 

atomdb/sql.py

@@ -1073,6 +1073,43 @@ def distinct(self, *args):
             distinct_clauses=distinct_clauses, related_clauses=related_clauses
         )
 
+    def where_clause(self, k: str, v: Any, related_clauses: ListType):
+        """ Create a where clause from a django-style parameter.
+        This will modify the list of related clauses if a join occurs.
+
+        Parameters
+        ----------
+        k: str
+            The filter key, eg name__startswith
+        v: object
+            The value
+        related_clauses: list
+            List of related clauses
+
+        Returns
+        -------
+        clause: sqlalchemy.sq.expression
+            The filter clause
+
+        """
+        model = self.proxy.model
+        op = "eq"
+        if "__" in k:
+            parts = k.split("__")
+            if parts[-1] in QUERY_OPS:
+                op = parts[-1]
+                k = "__".join(parts[:-1])
+        col = resolve_member_column(model, k, related_clauses)
+
+        # Support lookups by model
+        if isinstance(v, Model):
+            v = v.serializer.flatten_object(v, scope={})
+        elif op in ("in", "notin"):
+            # Flatten lists when using in or notin ops
+            v = model.serializer.flatten(v, scope={})
+
+        return getattr(col, QUERY_OPS[op])(v)
+
     def filter(self, *args, **kwargs: DictType[str, Any]):
         """Filter the query by the given parameters. This accepts sqlalchemy
         filters by arguments and django-style parameters as kwargs.
@@ -1094,30 +1131,53 @@ def filter(self, *args, **kwargs: DictType[str, Any]):
         filter_clauses = self.filter_clauses + list(args)
         related_clauses = self.related_clauses[:]
 
-        connection_kwarg, restore_kwarg = p.connection_kwarg, p.restore_kwarg
+        connection_kwarg = p.connection_kwarg
+        restore_kwarg = p.restore_kwarg
+
+        # Build the filter operations
+        for k, v in kwargs.items():
+            if k == connection_kwarg or k == restore_kwarg:
+                continue
+            filter_clauses.append(self.where_clause(k, v, related_clauses))
+
+        return self.clone(
+            connection=kwargs.get(connection_kwarg, self.connection),
+            force_restore=kwargs.get(restore_kwarg, self.force_restore),
+            filter_clauses=filter_clauses,
+            related_clauses=related_clauses,
+        )
+
+    def exclude(self, *args, **kwargs: DictType[str, Any]):
+        """Exclude results matching the given parameters by wrapping each
+        clause in a NOT expression. This accepts sqlalchemy filters by
+        arguments and django-style parameters as kwargs.
+
+        Parameters
+        ----------
+        args: List
+            List of sqlalchemy filters
+        kwargs: Dict[str, object]
+            Django style filters to use
+
+        Returns
+        -------
+        query: SQLQuerySet
+            A clone of this queryset with the excluded filter terms added.
+
+        """
+        p = self.proxy
+        filter_clauses = self.filter_clauses + [sa.not_(it) for it in args]
+        related_clauses = self.related_clauses[:]
+
+        connection_kwarg = p.connection_kwarg
+        restore_kwarg = p.restore_kwarg
 
         # Build the filter operations
         for k, v in kwargs.items():
             if k == connection_kwarg or k == restore_kwarg:
                 continue
-            model = p.model
-            op = "eq"
-            if "__" in k:
-                parts = k.split("__")
-                if parts[-1] in QUERY_OPS:
-                    op = parts[-1]
-                    k = "__".join(parts[:-1])
-            col = resolve_member_column(model, k, related_clauses)
-
-            # Support lookups by model
-            if isinstance(v, Model):
-                v = v.serializer.flatten_object(v, scope={})
-            elif op in ("in", "notin"):
-                # Flatten lists when using in or notin ops
-                v = model.serializer.flatten(v, scope={})
-
-            clause = getattr(col, QUERY_OPS[op])(v)
-            filter_clauses.append(clause)
+            clause = self.where_clause(k, v, related_clauses)
+            filter_clauses.append(sa.not_(clause))
 
         return self.clone(
             connection=kwargs.get(connection_kwarg, self.connection),

tests/test_sql.py

@@ -1036,17 +1036,17 @@ async def test_filters(db):
     await reset_tables(User)
 
     user, created = await User.objects.get_or_create(
-        name=faker.name(), email=faker.email(), age=21, active=True
+        name="Bob", email=faker.email(), age=21, active=True
     )
     assert created
 
     user2, created = await User.objects.get_or_create(
-        name=faker.name(), email=faker.email(), age=48, active=False, rating=10.0
+        name="Tom", email=faker.email(), age=48, active=False, rating=10.0
     )
     assert created
 
     # Startswith
-    u = await User.objects.get(name__startswith=user.name[0])
+    u = await User.objects.get(name__startswith="B")
     assert u.name == user.name
     assert u is user  # Now cached
 
@@ -1070,6 +1070,9 @@ async def test_filters(db):
     users = await User.objects.filter(age__lt=30)
     assert len(users) == 1 and users[0].age == user.age
 
+    users = await User.objects.exclude(age=21)
+    assert len(users) == 1 and users[0].age == 48
+
     # Not supported
     with pytest.raises(ValueError):
         users = await User.objects.filter(age__xor=1)
@@ -1083,6 +1086,24 @@ async def test_filters(db):
         users = await User.objects.filter(does_not_exist=True)
 
 
+async def test_filter_exclude(db):
+    await reset_tables(User)
+    # Create second user
+    await User.objects.create(name="Bob", email="[email protected]", age=40, active=True)
+    await User.objects.create(name="Jack", email="[email protected]", age=30, active=False)
+    await User.objects.create(name="Bob", email="[email protected]", age=20, active=False)
+
+    users = await User.objects.filter(name__startswith="B").exclude(
+        email__endswith="other.com")
+    assert len(users) == 1 and users[0].email == "[email protected]"
+
+    users = await User.objects.exclude(active=True, age__lt=25)
+    assert len(users) == 1 and users[0].name == "Jack"
+
+    users = await User.objects.exclude(name="Bob")
+    assert len(users) == 1 and users[0].name == "Jack"
+
+
 async def test_update(db):
     await reset_tables(User)
     # Create second user