Я работаю над мини-фреймворком на Python, предназначенным в основном для серверных REST API. Один из важных аспектов, над которым я сейчас работаю, — это тестирование, и я разрабатываю замену внутренней базы данных, которая переключает его с использования реальной базы данных на серверной части на использование хранилища данных в памяти. Цель состоит в том, чтобы упростить тестирование, поскольку вы можете запускать что-то более похожее на интеграционные тесты без необходимости предоставлять / настраивать фактическую базу данных.
Я не особо ориентирован на производительность, поэтому сейчас я просто храню записи в виде списка словарей. Сложная часть — это фильтрация. «Запросы» передаются от построителя запросов в виде словаря с wheres параметр. Это будет выглядеть примерно так:
{
"wheres": [
{"column": "age", "operator": "<", "values": [25]},
{"column": "status_id", "operator": "in", "values": [1, 2, 3]}
]
}
В случае бэкэнда курсора эти конфигурации обрабатываются и превращаются в подготовленный запрос. В случае бэкэнда API они превращаются в вызов API. В случае с памятью … я должен сам фильтровать. Мое решение для этого, вероятно, основано на более функциональных подходах JavaScript: у меня есть словарь с каждым поддерживаемым оператором в качестве ключа, а значение — это лямбда, которая возвращает лямбда, которая может использоваться для фильтрации. Рассматриваемый код ниже. Отметим, в частности, _operator_lambda_builders словарь и rows метод (внизу класса)
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)
Вот пример того, как вы могли бы использовать это на практике. Я «издевался» над реальной моделью (которая на самом деле не требуется для этого примера использования, но неявно требуется при текущем поведении класса):
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'}]
Для большего контекста вы обычно объявляете класс модели следующим образом:
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'),
])
Вы бы настроили правила внедрения зависимостей для своего теста, чтобы поменять местами cursor_backend для этого бэкэнда памяти. Это изменение прозрачно для модели, которую вы используете следующим образом:
user.create({'name': 'Bob', 'email': 'bob@example.com', 'age': 10})
И он передаст запрос на создание в MemoryBackend и оттуда к вышеизложенному MemoryTable класс.
Существует построитель запросов, который автоматически соединяет модели и серверную часть, а также генерирует конфигурацию для последующего вызова серверной части. Следовательно, если вы использовали такой конструктор запросов:
preschoolers = users.where('age<6').where('age>2')
for preschooler in preschooler:
print(preschooler.name)
Тогда MemoryTable рассматриваемый может увидеть, что это произойдет:
table = MemoryTable(model=user)
# records are added
table.rows({'wheres': [
{'column': 'age', 'operator': '<', 'values': [6]},
{'column': 'age', 'operator': '>', 'values': [2]}
]})
Я не могу решить, имеет ли построитель лямбда смысл, его невозможно прочитать или и то, и другое. Я также не могу решить, есть ли лучший подход. Код живет здесь и есть дополнительная документация в стадии разработки здесь.
Мне особенно любопытно, есть ли лучший, более читаемый подход (который не требует дополнительных сотен строк кода или бесконечного блока if / elif), но, конечно, я всегда готов к любому и все остальные предложения!
1 ответ
Если бы мне пришлось программировать ваш класс, я бы сосредоточился на определении операторов отдельно от таблицы. Желательно в классе, который выглядит как обычный код Python.
Если бы вы изменили операторов с = сказать, eq мы могли бы просто определить класс и использовать getattr. Или мы могли бы определить __getitem__ как удобство, позволяющее использовать тот же интерфейс, который вы используете.
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"]
Однако, поскольку ваши операторы являются недопустимыми символами в синтаксисе Python, нам понадобится способ определения пользовательских имен. Мы можем использовать декоратор для определения имен функций.
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) # =
Теперь мы можем сосредоточиться на том, чтобы Python добавил красивые имена в область видимости класса. Мы можем использовать __init_subclass__ установить методы с правильным именем в классе с setattr. Мы также можем добавить __getitem__ классу, чтобы разрешить доступ через [] синтаксис.
Наконец, у меня будет базовый класс и подкласс, так что если вам когда-нибудь понадобится определить другой набор операторов, вы можете это сделать.
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
Затем я бы немного изменил класс вашей таблицы, чтобы использовать новый класс.
def _where_as_filter(self, where):
column = where['column']
values = where['values']
return MemoryOperators(column, values)[where['operator']]
Это определенно интересный подход, который я не рассматривал
— Конор Манконе