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': '[email protected]', '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)
# records are added
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!