db.py

"""Data-access layer for HopeIT Manager.

The UI never imports pyodbc or writes SQL. It calls the functions here,
which open a connection from config.build_connection_string() (settings
come from config.ini), run
parameterised queries (always ? placeholders) and return plain Python
data structures (lists of dicts).

A failed database call raises DBError with a message that is safe to show
to the user; the technical detail goes to the caller only through that
message so the app can display it in a dialog without crashing.
"""

import pyodbc

import config


class DBError(Exception):
    """Raised for any database problem the UI should surface to the user."""


def _connect():
    try:
        return pyodbc.connect(
            config.build_connection_string(),
            timeout=config.CONNECTION_TIMEOUT,
        )
    except pyodbc.Error as exc:
        raise DBError(_friendly(exc)) from exc


def _friendly(exc: pyodbc.Error) -> str:
    """Turn a pyodbc error into a short readable sentence."""
    try:
        sqlstate = exc.args[0]
        detail = exc.args[1]
    except (IndexError, AttributeError):
        return str(exc)
    if sqlstate in ("28000", "18456"):
        return "Login failed. Check the username/password or use Windows Authentication in config.ini."
    if sqlstate in ("08001", "08S01", "HYT00"):
        return ("Cannot reach the SQL Server.\n"
                "Check server / DSN in config.ini and that the SQL Server service is running.")
    if "Cannot open database" in detail:
        return "The HopeIT database was not found. Run HopeIT_setup.sql first."
    return detail


def _rows(cursor):
    columns = [c[0] for c in cursor.description]
    return [dict(zip(columns, row)) for row in cursor.fetchall()]


def _run(sql, params=(), *, write=False):
    """Run one statement. Returns rows for reads, affected count for writes."""
    conn = _connect()
    try:
        cur = conn.cursor()
        cur.execute(sql, params)
        if write:
            affected = cur.rowcount
            conn.commit()
            return affected
        return _rows(cur)
    except pyodbc.IntegrityError as exc:
        conn.rollback()
        raise DBError(_integrity_message(exc)) from exc
    except pyodbc.Error as exc:
        conn.rollback()
        raise DBError(_friendly(exc)) from exc
    finally:
        conn.close()


def _integrity_message(exc: pyodbc.Error) -> str:
    text = str(exc)
    if "REFERENCE" in text or "FOREIGN KEY" in text:
        return "This record is used by other rows and cannot be deleted."
    if "CHECK constraint" in text:
        return "A value breaks a database rule (price/quantity must be positive)."
    if "duplicate" in text.lower() or "UNIQUE" in text:
        return "A record with that key already exists."
    return text


# --------------------------------------------------------------------- ping
def ping():
    """Return (ok, message). Never raises."""
    try:
        rows = _run("SELECT 1 AS ok;")
        if rows and rows[0]["ok"] == 1:
            return True, "Connected to HopeIT."
        return False, "Unexpected response from the database."
    except DBError as exc:
        return False, str(exc)


# ---------------------------------------------------------------- customers
def list_customers(search=None):
    if search:
        like = f"%{search}%"
        return _run(
            "SELECT CustomerID, CustomerName, City, Email, Phone "
            "FROM Customer "
            "WHERE CustomerName LIKE ? OR City LIKE ? "
            "ORDER BY CustomerName;",
            (like, like),
        )
    return _run(
        "SELECT CustomerID, CustomerName, City, Email, Phone "
        "FROM Customer ORDER BY CustomerName;"
    )


def add_customer(name, city, email, phone):
    return _run(
        "INSERT INTO Customer (CustomerName, City, Email, Phone) VALUES (?, ?, ?, ?);",
        (name, city, email, phone),
        write=True,
    )


def update_customer(customer_id, name, city, email, phone):
    return _run(
        "UPDATE Customer SET CustomerName = ?, City = ?, Email = ?, Phone = ? "
        "WHERE CustomerID = ?;",
        (name, city, email, phone, customer_id),
        write=True,
    )


def delete_customer(customer_id):
    return _run("DELETE FROM Customer WHERE CustomerID = ?;", (customer_id,), write=True)


# ----------------------------------------------------------------- products
def list_products(search=None):
    if search:
        like = f"%{search}%"
        return _run(
            "SELECT ProductID, ProductName, UnitPrice, StockQuantity "
            "FROM Products WHERE ProductName LIKE ? ORDER BY ProductName;",
            (like,),
        )
    return _run(
        "SELECT ProductID, ProductName, UnitPrice, StockQuantity "
        "FROM Products ORDER BY ProductName;"
    )


def add_product(name, unit_price, stock_quantity):
    return _run(
        "INSERT INTO Products (ProductName, UnitPrice, StockQuantity) VALUES (?, ?, ?);",
        (name, unit_price, stock_quantity),
        write=True,
    )


def update_product(product_id, name, unit_price, stock_quantity):
    return _run(
        "UPDATE Products SET ProductName = ?, UnitPrice = ?, StockQuantity = ? "
        "WHERE ProductID = ?;",
        (name, unit_price, stock_quantity, product_id),
        write=True,
    )


def delete_product(product_id):
    return _run("DELETE FROM Products WHERE ProductID = ?;", (product_id,), write=True)


# ------------------------------------------------------------------- orders
def list_customer_choices():
    return _run("SELECT CustomerID, CustomerName FROM Customer ORDER BY CustomerName;")


def list_product_choices():
    return _run(
        "SELECT ProductID, ProductName, UnitPrice FROM Products ORDER BY ProductName;"
    )


def create_order(customer_id, order_date, lines):
    """Insert one order and its detail lines in a single transaction.

    lines: iterable of (product_id, quantity).
    Returns the new OrderID.
    """
    if not lines:
        raise DBError("An order needs at least one line item.")
    conn = _connect()
    try:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO Orders (CustomerID, OrderDate) "
            "OUTPUT INSERTED.OrderID VALUES (?, ?);",
            (customer_id, order_date),
        )
        order_id = cur.fetchone()[0]
        for product_id, quantity in lines:
            cur.execute(
                "INSERT INTO OrderDetails (OrderID, ProductID, Quantity) VALUES (?, ?, ?);",
                (order_id, product_id, quantity),
            )
        conn.commit()
        return order_id
    except pyodbc.IntegrityError as exc:
        conn.rollback()
        raise DBError(_integrity_message(exc)) from exc
    except pyodbc.Error as exc:
        conn.rollback()
        raise DBError(_friendly(exc)) from exc
    finally:
        conn.close()


def list_orders():
    return _run(
        "SELECT o.OrderID, c.CustomerName, o.OrderDate, "
        "       COUNT(od.OrderDetailID)              AS LineItems, "
        "       ISNULL(SUM(od.Quantity * p.UnitPrice), 0) AS OrderTotal "
        "FROM Orders o "
        "JOIN Customer c        ON c.CustomerID = o.CustomerID "
        "LEFT JOIN OrderDetails od ON od.OrderID = o.OrderID "
        "LEFT JOIN Products p      ON p.ProductID = od.ProductID "
        "GROUP BY o.OrderID, c.CustomerName, o.OrderDate "
        "ORDER BY o.OrderID DESC;"
    )


def list_order_lines(order_id):
    return _run(
        "SELECT p.ProductName, od.Quantity, p.UnitPrice, "
        "       (od.Quantity * p.UnitPrice) AS LineTotal "
        "FROM OrderDetails od "
        "JOIN Products p ON p.ProductID = od.ProductID "
        "WHERE od.OrderID = ? "
        "ORDER BY p.ProductName;",
        (order_id,),
    )


# ------------------------------------------------------------------ reports
def report_customer_orders(customer_id=None):
    if customer_id:
        return _run(
            "SELECT c.CustomerName, o.OrderID, o.OrderDate, p.ProductName, "
            "       od.Quantity, p.UnitPrice, "
            "       (od.Quantity * p.UnitPrice) AS LineTotal "
            "FROM Customer c "
            "JOIN Orders o        ON o.CustomerID = c.CustomerID "
            "JOIN OrderDetails od ON od.OrderID   = o.OrderID "
            "JOIN Products p      ON p.ProductID  = od.ProductID "
            "WHERE c.CustomerID = ? "
            "ORDER BY o.OrderDate, o.OrderID;",
            (customer_id,),
        )
    return _run(
        "SELECT c.CustomerName, o.OrderID, o.OrderDate, p.ProductName, "
        "       od.Quantity, p.UnitPrice, "
        "       (od.Quantity * p.UnitPrice) AS LineTotal "
        "FROM Customer c "
        "JOIN Orders o        ON o.CustomerID = c.CustomerID "
        "JOIN OrderDetails od ON od.OrderID   = o.OrderID "
        "JOIN Products p      ON p.ProductID  = od.ProductID "
        "ORDER BY c.CustomerName, o.OrderDate, o.OrderID;"
    )


def report_sales_summary():
    return _run(
        "SELECT p.ProductName, "
        "       SUM(od.Quantity)               AS TotalQtySold, "
        "       SUM(od.Quantity * p.UnitPrice) AS TotalRevenue, "
        "       COUNT(DISTINCT od.OrderID)     AS OrdersContaining "
        "FROM Products p "
        "JOIN OrderDetails od ON od.ProductID = p.ProductID "
        "GROUP BY p.ProductName "
        "ORDER BY TotalRevenue DESC;"
    )


def report_low_stock(threshold):
    return _run(
        "SELECT ProductID, ProductName, UnitPrice, StockQuantity "
        "FROM Products WHERE StockQuantity < ? ORDER BY StockQuantity, ProductName;",
        (threshold,),
    )