应使用参数化查询配合上下文管理器或SQLAlchemy Core：一、sqlite3中用with connect()自动管理连接与事务，execute("WHERE age > ?", (25,))防注入；二、SQLAlchemy Core通过create_engine和text()实现数据库无关的参数化查询。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
如果您希望使用CodeGeeX辅助编写Python代码来执行数据库查询，但不确定如何以简洁、安全且符合Python惯用法的方式实现，则可能是由于缺乏对参数化查询、上下文管理及数据库驱动抽象层的统一理解。以下是实现该目标的多种具体方式：
一、使用sqlite3模块配合上下文管理器
SQLite是Python标准库内置的轻量级数据库，其connect()返回的对象支持上下文协议，可自动提交或回滚事务，并确保连接被正确关闭。采用with语句能避免资源泄漏，同时结合参数化查询防止SQL注入。
1、导入sqlite3模块：import sqlite3。
2、使用with语句连接数据库并创建游标：with sqlite3.connect("example.db") as conn: cursor = conn.cursor()。
立即学习“Python免费学习笔记（深入）”；
3、执行参数化查询语句：cursor.execute("SELECT * FROM users WHERE age > ?", (25,))。
4、获取全部结果：results = cursor.fetchall()。
二、使用SQLAlchemy Core构建可移植查询
SQLAlchemy Core提供数据库无关的表达式语言，通过text()或原生select()构造查询，既保持SQL灵活性，又支持绑定参数与方言适配。它不强制使用ORM模型，适合需要直接控制SQL逻辑的场景。
1、安装依赖：pip install sqlalchemy。
2、创建引擎并连接：from sqlalchemy import create_engine; engine = create_engine("sqlite:///example.db")。
3、定义参数化文本查询：from sqlalchemy import text; stmt = text("SELECT name FROM products WHERE price 。
4、执行并传入参数：with engine.connect() as conn: result = conn.execute(stmt, {"max_price": 100.0})。
5、遍历结果行：for row in result: print(row.name)。
三、使用asyncpg配合async/await异步查询PostgreSQL
当处理高并发I/O密集型数据库操作时，asyncpg提供极快的异步PostgreSQL访问能力，其查询接口默认启用二进制协议与预编译语句，性能优于同步驱动。所有查询必须在异步上下文中执行。
1、安装异步驱动：pip install asyncpg。
2、定义异步查询函数：import asyncpg; async def fetch_users(): ...。
3、建立异步连接池：pool = await asyncpg.create_pool("postgresql://user:pass@localhost/dbname")。
4、从池中获取连接并执行参数化查询：async with pool.acquire() as conn: rows = await conn.fetch("SELECT * FROM logs WHERE level = $1", "ERROR")。
5、解析结果为字典列表：[dict(row) for row in rows]。
四、使用Dataset库实现极简声明式查询
Dataset将数据库视为“表格集合”，隐藏连接、游标和SQL字符串拼接细节，允许以类似字典的操作方式读写数据。它内部仍使用SQLAlchemy引擎，适用于快速原型开发或脚本任务，但不适用于复杂JOIN或存储过程调用。
1、安装库：pip install dataset。
2、打开数据库连接：import dataset; db = dataset.connect("sqlite:///example.db")。
3、获取表对象：table = db["users"]。
4、执行带条件的查询：results = table.find(age={"$gt": 30}, order_by="-created_at", limit=10)。
5、迭代结果并访问字段：for record in results: print(record["name"], record["email"])。
