能否将SQLite3的conn.cursor()与commit()封装为Python函数？

• 缺少返回值：函数仅在内部完成SQL执行，既没有将查询得到的结果集向外传递，也没有返回增删改操作的影响行数，调用方执行完函数拿不到任何有效执行结果，自然判定功能异常
• 事务逻辑错位：对SELECT这类只读查询也强制执行conn.commit()，属于完全冗余的操作，还会产生无意义的磁盘写入开销
• 无异常兜底：SQL执行出错时没有配套的事务回滚逻辑，容易残留执行到一半的脏数据
• 硬依赖全局变量：代码中引用的conn如果未提前初始化、或是已经被关闭，会直接触发运行报错
• 缺少参数化支持：仅能接收拼接完成的纯SQL字符串，既存在SQL注入风险，复杂场景下手动拼接字符串也很容易触发语法错误

import sqlite3

# 提前完成数据库连接初始化，可按需替换为实际的库文件路径
conn = sqlite3.connect("local_test.db")
# 配置查询结果可按列名访问，比元组形式更易使用
conn.row_factory = sqlite3.Row

def sql(sql_str: str, params: tuple = (), need_fetch_all: bool = False):
    cursor = conn.cursor()
    try:
        cursor.execute(sql_str, params)
        # 非查询类操作才需要提交事务
        if not sql_str.strip().upper().startswith("SELECT"):
            conn.commit()
            return cursor.rowcount
        # 查询类操作按需求返回结果集
        return cursor.fetchall() if need_fetch_all else cursor.fetchone()
    except Exception:
        conn.rollback()  # 执行出错时回滚，避免脏数据残留
        raise

• 全量查询匹配数据：调用时传入need_fetch_all=True，即可拿到所有符合条件的行结果
  sql("SELECT * FROM user WHERE age >= ?", (18,), need_fetch_all=True)
• 查询单条匹配数据：直接调用即可返回第一条匹配的行结果
  sql("SELECT * FROM user WHERE id = ?", (1001,))
• 增/删/改操作：函数会自动完成事务提交，返回操作影响的行数
  sql("INSERT INTO user(name, age) VALUES (?, ?)", ("张三", 25))

注意：所有涉及动态变量的操作，统一通过params参数传值，不要直接把变量拼接到SQL字符串中。

内容的提问来源于stack exchange，提问作者mike