from __future__ import annotations import asyncpg class SimpleQueryBuilder: def __init__(self): self._table_name = "" self._select = ["*"] self._where = [] self._having = [] self._order_by = None self._order_by_desc = False def table(self, table_name: str): self._table_name = table_name return self def fields(self, fields: list[str]): self.select = fields return self def where(self, where: str, condition: str, param): self._where.append((where, condition, param)) return self def having(self, having: str, condition: str, param): self._having.append((having, condition, param)) return self def build(self): sql = "SELECT %s FROM %s" % (", ".join(self._select), self._table_name) params = [] paramsLen = 0 if len(self._where) > 0: sql += " WHERE " for where, condition, param in self._where: params.append(param) paramsLen += 1 sql += "%s %s $%d AND " % (where, condition, paramsLen) if self._order_by is not None: sql += " ORDER BY %s %s" % (self._order_by, "DESC" if self._order_by_desc else "ASC") if len(self._having) > 0: sql += " HAVING " for having, condition, param in self._having: params.append(param) paramsLen += 1 sql += "%s %s $%d AND " % (having, condition, paramsLen)