# In memory "table" filtering in Python

I'm working on a mini-framework in Python, designed largely for backend REST APIs. One important aspect I'm working on now is testing, and I'm designing a drop-in backend replacement that switches it from using an actual database on the backend to using an in-memory data store. The goal is to simplify testing since you can run something more akin to integration tests without having to provision/configure an actual database.

I'm not hyper focused on performance so right now this is just storing records as a list of dictionaries. The tricky part is filtering. "Queries" are passed in from the query builder as a dictionary with a wheres parameter. This will look something like this:

{
"wheres": [
{"column": "age", "operator": "<", "values": [25]},
{"column": "status_id", "operator": "in", "values": [1, 2, 3]}
]
}


In the case of the cursor backend these configurations are processed and turned into a prepared query. In the case of the API backend these are turned into an API call. In the case of the memory backend... I have to filter things myself. My solution for this is probably inspired by more functional approaches from JavaScript: I have a dictionary with each supported operator as a key, and the value is a lambda that returns a lambda that can be used for filtering. The code in question is below. Note in particular the _operator_lambda_builders dictionary and the rows method (at the bottom of the class)

class MemoryTable:
_table_name = None
_column_names = None
_rows = None
_id_to_index = None
_next_id = 1

# here be dragons.  This is not a 100% drop-in replacement for the equivalent SQL operators
_operator_lambda_builders = {
'<=>': lambda column, values: lambda row: (row[column] if column in row else None) == values[0],
'!=': lambda column, values: lambda row: (row[column] if column in row else None) != values[0],
'<=': lambda column, values: lambda row: (row[column] if column in row else None) <= values[0],
'>=': lambda column, values: lambda row: (row[column] if column in row else None) >= values[0],
'>': lambda column, values: lambda row: (row[column] if column in row else None) > values[0],
'<': lambda column, values: lambda row: (row[column] if column in row else None) < values[0],
'=': lambda column, values: lambda row: (row[column] if column in row else None) == values[0],
'is not null': lambda column, values: lambda row: (column in row and row[column] is not None),
'is null': lambda column, values: lambda row: (column not in row or row[column] is None),
'is not': lambda column, values: lambda row: (row[column] if column in row else None) != values[0],
'is': lambda column, values: lambda row: (row[column] if column in row else None) == values[0],
'like': lambda column, values: lambda row: (row[column] if column in row else None) == values[0],
'in': lambda column, values: lambda row: (row[column] if column in row else None) in values,
}

def __init__(self, model=None):
self._column_names = []
self._rows = []
self._id_to_index = {}

if model is not None:
self._table_name = model.table_name
self._column_names.extend(model.columns_configuration().keys())

def update(self, id, data):
if id not in self._id_to_index:
raise ValueError(f"Cannot update non existent record with id of '{id}'")
index = self._id_to_index[id]
if index is None:
raise ValueError(f"Cannot update record with id of '{id}' because it was already deleted")
for column_name in data.items():
if column_name not in self._column_names:
raise ValueError(
f"Cannot update record: column '{column_name}' does not exist in table '{self._table_name}'"
)
self._rows[index] = {
**self._rows[index],
**data,
}
return self._rows[index]

def create(self, data):
for column_name in data.keys():
if column_name not in self._column_names:
raise ValueError(
f"Cannot create record: column '{column_name}' does not exist in table '{self._table_name}'"
)
self._next_id += 1
new_id = self._next_id
data['id'] = new_id
for column_name in self._column_names:
if column_name not in data:
data[column_name] = None
self._rows.append(data)
self._id_to_index[new_id] = len(self._rows)-1
return self._rows

def delete(self, id):
if id not in self._id_to_index:
return
index = self._id_to_index[id]
if row_index is None:
return True
row = self._rows[row_index]
if row is None:
return True
# we set the row to None because if we remove it we'll change the indexes of the rest
# of the rows, and break our self._id_to_index map
self._rows[row_index] = None
self._id_to_index[id] = None

def count(self, configuration):
return len(self.rows(configuration))

def rows(self, configuration):
if 'wheres' in configuration:
rows = self._rows
for where in configuration['wheres']:
rows = filter(self._where_as_filter(where), rows)
rows = list(rows)
else:
rows = [*self._rows]
return rows

def _where_as_filter(self, where):
column = where['column']
values = where['values']
return self._operator_lambda_builders[where['operator']](column, values)


Here is an example of how you would use this in practice. I've "mocked" an actual model (which is not actually required for this example use case but is implicitly required with the current class behavior):

from clearskies.backends.memory_backend import MemoryTable
from types import SimpleNamespace

model = SimpleNamespace(table_name='people', columns_configuration=lambda: {'name': ''})
table = MemoryTable(model=model)
table.create({'name': 'alice'})
table.create({'name': 'bob'})
table.create({'name': 'jane'})
table.rows({'wheres': [{'column': 'name', 'operator': '=', 'values': ['bob']}]})

# returns [{'id': 2, 'name': 'bob'}]


For more context, you would normally declare a model class like this:

from collections import OrderedDict
from clearskies import Model
from clearskies.column_types import string, email, integer

class User(Model):
def __init__(self, cursor_backend, columns):
super().__init__(cursor_backend, columns)

def columns_configuration(self):
return OrderedDict([
string('name'),
email('email'),
integer('age'),
])


You would adjust the dependency injection rules for your test to swap out the cursor_backend for this memory backend. This change is transparent to the model, which you use like this:

user.create({'name': 'Bob', 'email': 'bob@example.com', 'age': 10})


And it would pass the create request down to the MemoryBackend and, from there, to the above MemoryTable class.

There is a query builder automatically connects models and the backend, as well as generate the configuration for the eventual backend call. Therefore, if you used a query builder like this:

preschoolers = users.where('age<6').where('age>2')
for preschooler in preschooler:
print(preschooler.name)


Then the MemoryTable in question would see this happen:

table = MemoryTable(model=user)
table.rows({'wheres': [
{'column': 'age', 'operator': '<', 'values': [6]},
{'column': 'age', 'operator': '>', 'values': [2]}
]})



I can't decide if the lambda builder makes perfect sense, is impossible to read, or both. I also can't decide if there is a better approach. The code lives here and there is further documentation under construction here.

I'm especially curious if there is a better, more readable approach (that doesn't make this take up a hundred extra lines of code or entail an endless if/elif block), but of course I'm always up for any and all other suggestions!

• When posting bounties filling out the optional box is a good idea. You can explain what you feel is lacking in the existing answers and helps you get the feedback you want. Since you chose not to fill in the box I'd recommend you edit your question. Apr 11 at 20:03
• @Peilonrayz there's nothing missing from your answer. I'm just curious if other people have other approaches/feedback Apr 12 at 9:23

If I were to program your class I'd focus on defining operators separately to the table. Preferably in a class which looks like normal Python code.

If you were to change your operators from = to, say, eq we could just define a class and use getattr. Or we could define __getitem__ as a convenience to allow the same interface you're using.

class MyOperators:
def __init__(self, column, values):
self.column = column
self.values = values

def __getitem__(self, key):
return getattr(self, key)

def eq(self, row):
return (row[self.column] if self.column in row else None) == self.values[0]

eq = MyOperators(column, values)["eq"]


However since your operators are invalid characters in Python's syntax we'd need a way to define custom names. We can use a decorator to define the names on the functions.

def operator(operator):
def inner(fn):
fn._operator = operator
return fn
return inner

@operator("=")
def eq(self, row):
return (row[self.column] if self.column in row else None) == self.values[0]

print(eq._operator)  # =


Now we can focus on getting Python to add the pretty names to the class' scope. We can use __init_subclass__ to set the methods with the correct name on the class with setattr. We can also add __getitem__ to the class to allow access through [] syntax.

Finally I'd have a base class and a subclass so if you ever need to define a different operator set you can.

def operator(operator):
def inner(fn):
fn._operator = operator
return fn
return inner

class Operators:
def __init_subclass__(cls):
for name in dir(cls):
fn = getattr(cls, name)
if hasattr(fn, "_operator"):
setattr(cls, fn._operator, fn)

def __getitem__(self, operator):
return getattr(self, operator)

class MemoryOperators(Operators):
def __init__(self, column, values):
self.column = column
self.values = values

@operator("=")
def eq(self, row):
return (row[self.column] if self.column in row else None) == self.values[0]

result = MemoryOperators("foo", ["bar"])["="]([{"foo": "bar"}])
print(result)  # True


I'd then change your table class slightly to use the new class.

def _where_as_filter(self, where):
column = where['column']
values = where['values']
return MemoryOperators(column, values)[where['operator']]

• That's definitely an interesting approach I had not considered Apr 7 at 17:47

Some quick observations, mostly off the top of my head and untested:

### dict.get()

dict.get(key) returns None if the key isn't in the dict. In the lambda functions,

(row[column] if column in row else None)


can be replaced with

row.get(column)


### None isn't orderable

However, there is a potential bug in many of the lambda functions. In Python 3.x, None < 6 raises a TypeError: '<' not supported between instances of 'NoneType' and 'int'. Same for any other numeric types and relative comparison operators. So if a column is missing, or the column value is None, TypeError will be raised.

Using None for null in the database feels prone to bugs. I'd suggest creating a class Null that defines the comparison operators appropriately. Then create a singleton instance, DBNULL, and use that in the database.

class Null:
def __lt__(self, other):
return False

def __eq__(self, other):
return isinstance(other, Null)

...

DBNULL = Null()


The lambda functions would then contain

row.get(column, DBNULL)


### '_next_idand_id_to_index

Based on the code provided, the handling of id's and indexes seem overly complicated. Just let the id be the index (or maybe index + 1 if you don't want 0 for an id).

def create(self, data):
...
data['id'] = len(self._rows)   # maybe + 1
...


Alternatively, let _rows be a dict keyed by id.

def create(self, data):
...
data['id'] = id = len(self._rows)   # maybe + 1
self._rows[id] = data
...


### dict.keys() returns a key view

A key view, such as returned by dict.keys(), is set like and supports set operations. set.subtract() can take an iterable. So

    for column_name in data.keys():
if column_name not in self._column_names:
raise ValueError( ... )


can be

    unknown_columns = data.keys().subtract(self._column_names)
if unknown_columns:
raise ValueError( ... )


or using the walrus operator:

    if (unknown_columns := data.keys().subtract(self._column_names))
raise ValueError( ... )


### ownership of data

Method create() only stores a reference to data in self._rows(). It does not make a copy. This is a potential source of hard to find bugs. Changes made to data by the calling code will be reflected in the database. For example:

table = MemoryTable(model=model)
d = {'name': 'alice'}
table.create(d)
d['name'] = 'bob'   # <<< this changes the value in the previous database record
table.create(d)


Either the caller needs to create new dicts each time or create should make a copy.

### dict.from_keys()

To create a new dict with all keys filled in, you could use:

    new_row = dict.from_keys(self._column_names, DBNULL)
new_row.update(data)


    for column_name in self._column_names:
if column_name not in data:
data[column_name] = None


### A list comprehension vs filter()

The nested lambdas in _operator_lambda_builders seem to needlessly complicate things. A list comprehension seems simpler:

def rows(self, configuration):
if 'wheres' in configuration:
rows = self._rows
for where in configuration['wheres']:
column = where['column']
values = where['values']
op = _operator_funcs[where['operator']]
rows = [row for row in rows if op(row, column, values)]
else:
rows = [*self._rows]
return rows


where:

_operator_funcs = {
...
'<':lambda row, col, vals: row.get(col, DBNULL) < vals[0],
...
}


## An alternative for incremented values

You have an incremented ID that is declared like this:

_next_id = 1


And then in the create routine it is incremented like this:

self._next_id += 1
new_id = self._next_id


While there's nothing wrong with your approach, I would suggest itertools.count instead.

import itertools

counter = itertools.count(start=1)


And then:

self.next_id = next(counter)


I find that approach slightly safer.

To quote the function in full:

def create(self, data):
for column_name in data.keys():
if column_name not in self._column_names:
raise ValueError(
f"Cannot create record: column '{column_name}' does not exist in table '{self._table_name}'"
)
self._next_id += 1
new_id = self._next_id
data['id'] = new_id
for column_name in self._column_names:
if column_name not in data:
data[column_name] = None
self._rows.append(data)
self._id_to_index[new_id] = len(self._rows)-1
return self._rows


At line 61:

if column_name not in data:


Did you mean:

if column_name not in data.keys():


That would be consistent with line 52:

for column_name in data.keys():


## Return values

Your function delete returns a boolean (True) at two places, except on line 69 where it simply is: return

A function should behave consistently and always return a predictable data type. So I suggest that you return a boolean value everywhere. Including at the end of your function.

I would even redeclare it like this with type hinting:

def delete(self, id: int) -> bool:


to make it clear how it works just by looking at the declaration. Although it is perfectly possible that you don't actually check the return value for this function, because it is always expected to succeed. But it seems to me that this condition:

if id not in self._id_to_index:
return


should return False, and be handled accordingly. If you're trying to delete a value that is not there anymore, it's no big deal but this could a sign of a bug somewhere in your program. Even if the delete is user-initiated it is still good to make the difference between success and failure obvious.

In that delete function, at line 70:

    index = self._id_to_index[id]
if row_index is None:


index is assigned but not used it seems. Where does row_index come from ?

## Indexing

IMO this is the weakness in your system. My recommendation is to ditch _id_to_index, you can do without it. I would rethink the approach. In the create function you have this: data['id'] = new_id so we assume the ID is properly incremented and unique throughout execution of your program. So all you need is a list comprehension to locate an entry in a list of dict, and update (or delete) that entry as desired. Something like:

row = [item for item in rows if item["id"] == 3]


will return one row, or [] if the ID does not exist. So the length of row shall be 0 or 1.

And note that to make updating easier you can even retrieve the current index in the list using: rows.index(row) since it is possible to find the index of a list item by value. Unless there are pitfalls I have not considered, it seems that locating a full dict in a list works fine. But have a look here too: Find the index of a dict within a list, by matching the dict's value. You could as well have a dict of dict instead of a list of dict.

Since you are using functions to update or delete rows by ID, you can easily afford this small abstraction. I still expect performance to be good.

• Great points, thank you! I especially like the itertools approach for ids - I wasn't aware of that. I think you make great points on indexing. Originally I planned on having actual` indexes, but decided that performance wasn't a focus in this code and it wasn't worth taking the time to build. If that's the case though, why bother making an index for the id? This isn't intended for large tables, so performance is unlikely to matter anyway. Apr 12 at 9:27