DAO类中直接拼接SQL字符串会导致SQL注入风险、语法错误、参数错位、字段混淆及维护困难；应使用参数化查询、Query Object分层、字典游标和dataclass映射来提升安全性、可读性与可维护性。
DAO类里直接拼接SQL字符串会出什么问题
硬编码SQL字符串到Python方法里，很快就会失控：字段增减、表关联变化、WHERE条件动态调整时，每次都要改字符串拼接逻辑。更麻烦的是SQL注入风险——哪怕用了format()或%，只要变量没经过严格过滤，user_input混进WHERE name = '{name}'这种写法，就等于给攻击者开了门。
实际开发中常见错误现象包括：mysql.connector.errors.ProgrammingError: 1064 (42000)（语法错）、tuple index out of range（参数数量不匹配）、以及生产环境突然查不到数据但日志里SQL看着“好像没错”。
- 用
str.format()或%拼接，参数顺序稍有变动就崩 - 多表JOIN时字段别名冲突，比如两个表都有
id，cursor.fetchall()返回的元组里分不清哪个是哪个 - WHERE条件带
IN子句时，传入列表长度不确定，无法预设占位符数量
用MySQL Connector/Python的parameterized query替代字符串拼接
MySQL官方驱动支持%s占位符，底层自动转义、类型适配，且能复用预编译语句（prepared statement），性能和安全都靠得住。关键不是“能不能用”，而是怎么组织DAO方法让参数传递清晰、可读、可测。
例如查询用户及其订单总数，不要写"SELECT u.name, COUNT(o.id) FROM user u LEFT JOIN order o ON u.id=o.user_id WHERE u.status=%s GROUP BY u.id"然后靠注释说明%s对应status，而应明确拆解：
立即学习“Python免费学习笔记（深入）”；
def get_user_with_order_count(self, status: str, min_orders: int = 0) -> list:
sql = """
SELECT u.id, u.name, COUNT(o.id) AS order_count
FROM user u
LEFT JOIN `order` o ON u.id = o.user_id
WHERE u.status = %s
GROUP BY u.id
HAVING COUNT(o.id) >= %s
"""
return self._execute_query(sql, (status, min_orders))注意：`order`加反引号是因为order是MySQL保留字；HAVING不能写成WHERE，否则过滤失效；参数必须是tuple或list，单个值也要写成(status,)，少个逗号就报错。
复杂查询需要分层：DAO只管执行，逻辑交给Service或Query Object
DAO类职责必须窄——只做连接管理、SQL执行、结果映射。把“查最近7天活跃用户且下单金额超500的VIP”这种业务规则塞进DAO方法名或参数里，等于把业务逻辑和数据访问耦合死。下次运营要改成“最近15天+金额超800”，你就得改DAO、测DAO、发版DAO。
推荐做法是引入轻量级Query Object：
class UserQuery:
def __init__(self):
self.status = 'active'
self.days = 7
self.min_amount = 500
<pre class='brush:python;toolbar:false;'>def to_sql(self) -> str:
return f"""
SELECT u.* FROM user u
INNER JOIN (
SELECT user_id FROM `order`
WHERE created_at >= DATE_SUB(NOW(), INTERVAL %s DAY)
GROUP BY user_id
HAVING SUM(amount) >= %s
) o ON u.id = o.user_id
WHERE u.status = %s
"""
def to_params(self) -> tuple:
return (self.days, self.min_amount, self.status)</pre>DAO里只接收这个对象：
def find_by_query(self, query: UserQuery) -> list:
return self._execute_query(query.to_sql(), query.to_params())这样DAO不变，新增查询只需新增Query类，老逻辑不受影响，单元测试也容易mock。
结果集映射别依赖索引序号，优先用dict或dataclass
用cursor.fetchone()返回tuple，靠row[0]、row[1]取值，一旦SQL里SELECT字段顺序调整或增删，所有调用处全崩。尤其当查询包含COALESCE、CASE WHEN或子查询时，字段顺序更难维护。
正确做法是启用字典游标（dict cursor）：
# 初始化连接时指定
self.connection = mysql.connector.connect(
...,
cursorclass=mysql.connector.cursor.MySQLCursorDict
)之后cursor.fetchall()返回的就是[{'id': 1, 'name': 'Alice', 'order_count': 5}, ...]，字段名即键名，和SQL里AS别名完全一致。如果用PyMySQL，对应的是DictCursor；SQLAlchemy则天然支持result.mappings()。
若需强类型校验，用dataclass配合**row解包：
@dataclass
class UserWithOrderCount:
id: int
name: str
order_count: int
<h1>在DAO方法内</h1><p>rows = self._execute_query(sql, params)
return [UserWithOrderCount(**r) for r in rows]字段名拼错或类型不匹配，运行时立刻报错，比后期查半天None值靠谱得多。
真正麻烦的从来不是写DAO，而是当一个查询同时涉及5张表、3层嵌套子查询、多种动态过滤条件时，如何让后续接手的人不用翻三天代码才能看懂WHERE里那个AND (u.type IN (...) OR u.created_at > ...)到底在筛什么。这时候Query Object + 显式参数 + 字典结果，就是最不聪明但最稳的解法。
