WIP: Mini queryset

Author: tony

libvcs/projects/git.py

@@ -15,10 +15,11 @@
     - [`GitProject.get_git_version`](libvcs.git.GitProject.get_git_version)
 """  # NOQA: E501
 import dataclasses
+import functools
 import logging
 import pathlib
 import re
-from typing import Dict, Optional, TypedDict, Union
+from typing import Any, Dict, Generic, Optional, TypedDict, TypeVar, Union
 from urllib import parse as urlparse
 
 from .. import exc
@@ -27,6 +28,106 @@
 logger = logging.getLogger(__name__)
 
 
+T = TypeVar("T", Any, Any)
+
+
+def parse_lookup(path, lookup):
+    """mykey__endswith("mykey") -> "mykey" else None"""
+    try:
+        return path.endswith(lookup) and path.rsplit(lookup)[0]
+    except Exception:
+        return None
+
+
+@functools.total_ordering
+@dataclasses.dataclass(eq=False)
+class ListQuery(Generic[T]):
+    data: list[T]
+
+    def __eq__(self, other):
+        return set(self.data) == set(other.data)
+
+    def __lt__(self, other):
+        return set(self.data).issubset(set(other.data))
+
+    def filter(self, **kwargs):
+        if len(kwargs.values()) == 0:
+            return self
+
+        def _filter(obj) -> bool:
+            for query_lookup, v in kwargs.items():
+                if (
+                    field := obj.get(parse_lookup(query_lookup, "__contains"), None)
+                ) is not None:
+                    if not field.contains(v):
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__icontains"), None)
+                ) is not None:
+                    if not field.lower().contains(v.lower()):
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__in"), None)
+                ) is not None:
+                    if field not in v:
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__nin"), None)
+                ) is not None:
+                    if field in v:
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__startswith"), None)
+                ) is not None:
+                    if not field.startswith(v):
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__istartswith"), None)
+                ) is not None:
+                    if not obj.lower().startswith(v.lower()):
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__endswith"), None)
+                ) is not None:
+                    if not field.endswith(v):
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__iendswith"), None)
+                ) is not None:
+                    if not field.lower().endswith(v.lower()):
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__regex"), None)
+                ) is not None:
+                    if not re.search(field, v):
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__iregex"), None)
+                ) is not None:
+                    if not re.search(field, v, re.IGNORECASE):
+                        return false
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__iexact"), None)
+                ) is not None:
+                    if not field.lower() != v.lower():
+                        return False
+                elif (
+                    field := obj.get(parse_lookup(query_lookup, "__exact"), None)
+                ) is not None:  # same as else
+                    if field != obj:
+                        return False
+                else:
+                    if (
+                        field := obj.get(parse_lookup(query_lookup, "__exact"), None)
+                    ) is not None:  # same as else
+                        if v != obj:
+                            return False
+
+            return True
+
+        return dataclasses.replace(self, data=[k for k in self.data if _filter(k)])
+
+
 class GitRemoteDict(TypedDict):
     """For use when hydrating GitProject via dict."""
 

tests/projects/test_git.py

@@ -3,7 +3,7 @@
 import os
 import pathlib
 import textwrap
-from typing import Callable
+from typing import Callable, Optional
 
 import pytest
 
@@ -17,6 +17,7 @@
     GitProject,
     GitRemote,
     GitStatus,
+    ListQuery,
     convert_pip_url as git_convert_pip_url,
 )
 from libvcs.shortcuts import create_project_from_pip_url
@@ -747,3 +748,18 @@ def test_repo_git_remote_checkout(
 
     assert git_repo_checkout_dir.exists()
     assert pathlib.Path(git_repo_checkout_dir / ".git").exists()
+
+
+@pytest.mark.parametrize(
+    "items,filter_expr,expected_result",
+    [
+        [[dict(apple="fruit")], None, ListQuery([dict(apple="fruit")])],
+        [[1, 2, 3, 4, 5], None, ListQuery([1, 2, 3, 4, 5])],
+    ],
+)
+def test_ListQuery(items: list, filter_expr: Optional[dict], expected_result: list):
+    qs = ListQuery(data=items)
+    if filter_expr is not None:
+        assert qs.filter(**filter_expr) == expected_result
+    else:
+        assert qs.filter() == expected_result