Placeholder translator for use with asyncpg

asyncpg_ph_translator.py

import logging
import re

_ph_catcher = re.compile('\\{(\\w+)\\}')


def translate_placeholders(query: str, params: Dict[str, Any]) -> Tuple[str, List[Any]]:
    """
    This is a simple translator designed to allow usage of queries like
    `SELECT a, b, c FROM table WHERE d = {d} AND e = {e}` with parameters in
    form of `{'d': 1, 'e': 2}`.

    It converts query to native PostgreSQL placeholder syntax
    `SELECT a, b, c FROM table WHERE d = $1 AND e = $2` and parameters to `[1, 2]`
    preserving arguments order and filtering out unused parameter values.

    :param query: str query
    :param params: dict params

    :return: tuple (translated query, [id])
    """
    mapping = {}
    values = []
    used_keys = list(m.group(1) for m in _ph_catcher.finditer(query))

    logging.log(logging.DEBUG - 5, 'Translating query: %s', query)
    logging.log(logging.DEBUG - 5, 'With params: %r', params)

    for k in params:
        if k not in used_keys:
            continue
        values.append(params[k])
        mapping[k] = f'${len(values)}'

    logging.log(logging.DEBUG - 5, 'Built mapping: %s', mapping)

    query = query.format_map(mapping)
    logging.debug('Query translated: %s', query)
    logging.debug('Query params translated: %r', values)

    return query, values