Parameterized SQLite Queries

from __future__ import annotations

import sqlite3
from contextlib import closing
from typing import Sequence


ALLOWED_SORTS = {"id": "id", "name": "name", "price": "price"}


class RepositoryError(RuntimeError):
    """Raised when a database operation cannot be completed."""


class ProductRepository:
    def __init__(self, connection: sqlite3.Connection) -> None:
        self._connection = connection
        self._connection.row_factory = sqlite3.Row
        self._connection.execute("PRAGMA foreign_keys = ON")

    def initialize(self) -> None:
        with self._connection:
            self._connection.execute(
                """
                CREATE TABLE IF NOT EXISTS products (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL,
                    price REAL NOT NULL CHECK (price >= 0)
                )
                """
            )

    def add(self, name: str, price: float) -> int:
        if not isinstance(name, str) or not name.strip():
            raise ValueError("name must be a non-empty string")
        if (
            not isinstance(price, (int, float))
            or isinstance(price, bool)
            or price < 0
        ):
            raise ValueError("price must be a non-negative number")

        try:
            with self._connection:
                cursor = self._connection.execute(
                    "INSERT INTO products (name, price) VALUES (?, ?)",
                    (name.strip(), float(price)),
                )
        except sqlite3.Error as error:
            raise RepositoryError(f"insert failed: {error}") from error

        if cursor.lastrowid is None:
            raise RepositoryError("insert did not return a row id")
        return int(cursor.lastrowid)

    def search(
        self,
        *,
        min_price: float = 0.0,
        names: Sequence[str] | None = None,
        sort: str = "price",
        descending: bool = False,
        limit: int = 20,
    ) -> list[sqlite3.Row]:
        if sort not in ALLOWED_SORTS:
            raise RepositoryError(f"unsupported sort column: {sort}")
        if not isinstance(min_price, (int, float)) or isinstance(min_price, bool):
            raise TypeError("min_price must be a number")
        if min_price < 0:
            raise ValueError("min_price must be non-negative")
        if not isinstance(limit, int) or not 1 <= limit <= 100:
            raise ValueError("limit must be between 1 and 100")
        if isinstance(names, str):
            raise TypeError("names must be a sequence of strings")

        sql = "SELECT id, name, price FROM products WHERE price >= ?"
        parameters: list[object] = [float(min_price)]

        if names is not None:
            cleaned = [name for name in names if isinstance(name, str) and name]
            if not cleaned:
                raise ValueError("names must contain a non-empty string")
            placeholders = ", ".join("?" for _ in cleaned)
            sql += f" AND name IN ({placeholders})"
            parameters.extend(cleaned)

        direction = "DESC" if descending else "ASC"
        sql += f" ORDER BY {ALLOWED_SORTS[sort]} {direction} LIMIT ?"
        parameters.append(limit)

        try:
            rows = self._connection.execute(sql, parameters).fetchall()
        except sqlite3.Error as error:
            raise RepositoryError(f"query failed: {error}") from error
        return list(rows)


def main() -> None:
    with closing(sqlite3.connect(":memory:")) as connection:
        repository = ProductRepository(connection)
        repository.initialize()
        for name, price in [
            ("keyboard", 49.9),
            ("monitor", 219.0),
            ("mouse", 25.5),
            ("USB hub", 39.0),
        ]:
            repository.add(name, price)

        rows = repository.search(
            names=["keyboard", "mouse"],
            sort="price",
            descending=True,
        )
        print([dict(row) for row in rows])
        print([dict(row) for row in repository.search(min_price=100)])

        try:
            repository.search(sort="price; DROP TABLE products")
        except RepositoryError as error:
            print(error)

        count = connection.execute("SELECT COUNT(*) FROM products").fetchone()[0]
        print(f"rows after rejected sort: {count}")


if __name__ == "__main__":
    main()