|
import requests |
|
|
|
_OPERATORS = { |
|
"$eq": lambda x, y: x == y, |
|
"$neq": lambda x, y: x != y, |
|
"$lt": lambda x, y: x < y, |
|
"$lte": lambda x, y: x <= y, |
|
"$gt": lambda x, y: x > y, |
|
"$gte": lambda x, y: x >= y, |
|
"$in": lambda x, y: x in y, |
|
"$nin": lambda x, y: x not in y, |
|
} |
|
|
|
|
|
class RemoteDataComparator: |
|
|
|
def __init__(self, k, op, v): |
|
self.k = k |
|
self._OP = _OPERATORS[op] |
|
self.v = v |
|
|
|
def evaluate(self, item) -> bool: |
|
return self._OP(item[self.k], self.v) |
|
|
|
def __rand__(self, other: RemoteDataComparator): |
|
# Must return new RemoteDataComparator that is a bitwise AND |
|
raise NotImplementedError() |
|
|
|
|
|
class RemoteDataField: |
|
|
|
def __init__(self, key: str): |
|
self.name = key |
|
|
|
def __eq__(self, other): |
|
return RemoteDataComparator(self.name, "$eq", other) |
|
|
|
def __ne__(self, other): |
|
return RemoteDataComparator(self.name, "$neq", other) |
|
|
|
|
|
class RemoteDataProvider: |
|
|
|
def __init__(self, list_all, get_one): |
|
self._list_all = list_all |
|
self._get_one = get_one |
|
self._data = None |
|
|
|
def __getattr__(self, key: str): |
|
return RemoteDataField(key) |
|
|
|
def __getitem__(self, ind: tuple): |
|
if isinstance(ind, RemoteDataComparator): |
|
return self._filter_where(ind) |
|
return self._data_at_index(ind) |
|
|
|
def _data_at_index(self, ind): |
|
return self._data[ind] |
|
|
|
def all_data(self): |
|
if not self._data: |
|
self._data = self._list_all() |
|
return self._data |
|
|
|
def _filter_where(self, ind): |
|
# Naïve approach: just get all, and then filter |
|
return [item for item in self.all_data() if ind.evaluate(item)] |
|
|