BrandCultivation/docs/superpowers/specs/2026-03-15-database-bugfix-design.md

Database Bug Fix Design

Date: 2026-03-15 Scope: database/db/mysql.py, database/dao/mysql_dao.py Approach: Surgical bug fixes only (no architectural changes)

Problems

mysql.py

1. connect_database: try/except only wraps connection string construction, not create_engine. Real connection failures are uncaught.
2. load_data_with_page: Uses query.replace("SELECT *", "SELECT COUNT(*)") to build count query. Breaks if query doesn't start with SELECT *. Also mutates the caller's params dict in-place by adding limit and offset keys.
3. fetch_all / fetch_one: Catch SQLAlchemyError, print, and return None silently. Callers cannot detect failures.

mysql_dao.py

1. get_cust_by_ids, get_shop_by_ids, get_product_by_ids, get_order_by_product_ids: Build IN (...) clause via string concatenation — SQL injection risk.
2. get_product_by_id: Queries a single record using load_data_with_page (pagination overhead for one row).
3. get_cust_list, get_product_from_order: Call fetch_all(text(query)) directly while all other methods use load_data_with_page — inconsistent interface. Depends on Fix 2 being applied first.

Fixes

mysql.py

Fix 1 — connect_database Move create_engine(...) inside the try block so connection failures are caught and re-raised as ConnectionAbortedError.

Fix 2 — load_data_with_page Replace string substitution with subquery wrapping:

SELECT COUNT(*) FROM (<original_query>) AS _count_subq

Works regardless of the original SELECT clause. Also copy params before mutating to avoid side-effects on the caller's dict:

params = dict(params)  # avoid mutating caller's dict

Fix 3 — fetch_all / fetch_one Add raise in the except block after printing the error.

Note: This is a behavior-breaking change. Currently callers receive None on failure; after this fix they receive an unhandled exception. This is intentional — silent failures are harder to debug than explicit ones. Callers that currently rely on None returns (e.g. pd.DataFrame(self.db_helper.fetch_all(...))) will propagate the exception upward, which is the correct behavior.

mysql_dao.py

Fix 4 — SQL injection in IN clauses load_data_with_page processes its query argument as a plain string (string concatenation, then wraps in text()), so pre-built text() objects with bindparam(expanding=True) are incompatible with it.

Instead, switch the four affected methods to use fetch_all directly with bindparam(expanding=True):

from sqlalchemy import bindparam, text

query = text("""
    SELECT * FROM table
    WHERE city_uuid = :city_uuid
    AND col IN :ids
""").bindparams(bindparam("ids", expanding=True))
params = {"city_uuid": city_uuid, "ids": list(id_list)}
data = pd.DataFrame(self.db_helper.fetch_all(query, params))

These methods are bounded by the input list size, so skipping pagination is safe.

Fix 5 — get_product_by_id Replace load_data_with_page with fetch_one. Wrap the result to return a single-row DataFrame consistent with other methods:

result = self.db_helper.fetch_one(text(query), params)
return pd.DataFrame([dict(result._mapping)] if result else [])

Fix 6 — get_cust_list / get_product_from_order Replace direct fetch_all(text(query)) calls with load_data_with_page(query, params). Requires Fix 2 to be applied first (subquery count approach handles SELECT DISTINCT correctly).

Fix Application Order

Fixes must be applied in this order due to dependencies:

1. Fix 2 (load_data_with_page subquery count)
2. Fix 1, Fix 3 (independent, can be done together)
3. Fix 6 (depends on Fix 2)
4. Fix 4, Fix 5 (independent of above)

Out of Scope

• No architectural changes (no read/write separation, no ORM models)
• No changes to Redis layer
• No changes to DAO method signatures
• No additional error handling in DAO callers