Query Builder Class — Chainable, Reusable SQL in Python

A query builder wraps SQL construction in a fluent Python API. Each method returns self so calls can be chained. The final .build() returns a safe (sql, params) tuple ready for psycopg2.

from __future__ import annotations
from typing import Any
import psycopg2
from psycopg2.extras import RealDictCursor


class QueryBuilder:
    """Minimal, safe SQL SELECT builder for PostgreSQL."""

    def __init__(self, table: str):
        self._table      = table
        self._selects    = ["*"]
        self._joins      = []
        self._wheres     = []
        self._params     = []
        self._group_bys  = []
        self._havings    = []
        self._havparams  = []
        self._order_bys  = []
        self._limit_val  = None
        self._offset_val = None

    # ── Builder methods ───────────────────────────────────────────
    def select(self, *cols: str) -> QueryBuilder:
        self._selects = list(cols)
        return self

    def join(self, clause: str) -> QueryBuilder:
        self._joins.append(clause)
        return self

    def where(self, clause: str, *params: Any) -> QueryBuilder:
        self._wheres.append(clause)
        self._params.extend(params)
        return self

    def group_by(self, *cols: str) -> QueryBuilder:
        self._group_bys.extend(cols)
        return self

    def having(self, clause: str, *params: Any) -> QueryBuilder:
        self._havings.append(clause)
        self._havparams.extend(params)
        return self

    def order_by(self, col: str, direction: str = "DESC") -> QueryBuilder:
        safe_dir = "DESC" if direction.upper() == "DESC" else "ASC"
        self._order_bys.append(f"{col} {safe_dir}")
        return self

    def limit(self, n: int) -> QueryBuilder:
        self._limit_val = n
        return self

    def offset(self, n: int) -> QueryBuilder:
        self._offset_val = n
        return self

    def build(self) -> tuple[str, list]:
        parts = [f"SELECT {', '.join(self._selects)}",
                 f"FROM {self._table}"]
        parts.extend(self._joins)
        if self._wheres:
            parts.append("WHERE " + " AND ".join(self._wheres))
        if self._group_bys:
            parts.append("GROUP BY " + ", ".join(self._group_bys))
        if self._havings:
            parts.append("HAVING " + " AND ".join(self._havings))
        if self._order_bys:
            parts.append("ORDER BY " + ", ".join(self._order_bys))

        all_params = list(self._params) + list(self._havparams)
        if self._limit_val is not None:
            parts.append("LIMIT %s")
            all_params.append(self._limit_val)
        if self._offset_val is not None:
            parts.append("OFFSET %s")
            all_params.append(self._offset_val)

        return "
".join(parts), all_params

    def execute(self, cur) -> list[dict]:
        sql, params = self.build()
        cur.execute(sql, params)
        return cur.fetchall()