Dynamic Binding for PostgreSQL with AsyncPG #python #postgresql #asyncpg

sqlparams.py

from collections import OrderdDict
from typings import Hashable, List


class SQLParams:

    def __init__(self):
        self.__values = OrderedDict()
        self.__count = 1

    def bind(self, value: Hashable):
        index = self.__values.get(value)
        if index is None:
            index = self.__count
            self.__values[value] = index
            self.__count += 1
        return f"${index}"

    def bindings(self) -> List[Hashable]:
        return list(self.__values.keys())

Use it like conn.fetchrow(query, *params.bindings()).

Useful when constructing queries with dynamic number of parameters.

Avoid sql interpolation or concatenation with user provided parameters. Always try to do binding!!!

But this won't work for table names... etc. In those cases you have to do a direct conversion to a string enum!