Здравствуйте, я пишу SQL запросы, используя библиотеку asyncpg и несколько методов у меня получились несколько однообразными. Я никак не могу найти способ сделать так, чтобы исключить некоторое дублирование в них.
Вот они:
async def insert(table_name: str, data: dict) -> str:
columns, values = data.keys(), data.values()
joined_values = ', '.join(values)
placeholders = ['$' + str(index) for index in range(1, len(columns) + 1)]
query = 'INSERT INTO %s (%s) VALUES (%s)' % \
(
table_name,
', '.join(columns),
', '.join(placeholders)
)
return await execute_with_transaction(query, joined_values)
async def upsert(table_name: str, data: dict, column: str, value: Any):
columns, values = data.keys(), data.values()
joined_values = ', '.join(values)
permissible_placeholder = len(columns) + 1
placeholders = []
for index, column in enumerate(columns, start=1):
placeholder = column + '=$' + index
placeholders.append(placeholder)
query = 'UPDATE %s SET %s WHERE %s=$%s' % \
(
table_name,
placeholders,
column,
permissible_placeholder
)
return await execute_with_transaction(query, joined_values, value)
Здесь функции принимают в качестве аргументов словарь, ключи которого являются колонками, далее все эти данные джоинятся для того, чтобы приобрести вид, пригодный для генерации SQL запроса и собираются плейсхолдеры для передачи данных из вне.
Эти методы получились очень однообразными и в них есть очень много дублирующихся действий. Возможно вы подскажете мне решение, способное их привести к общему знаменателю? Возможно декоратор какой-то?