SQL字符串拼接在异步任务中特别危险，因其不仅引发SQL注入（如f"SELECT * FROM users WHERE id = {user_id}"），且错误更隐蔽：恶意查询混于并发协程中，仅报OperationalError等模糊异常；若误用同步驱动（如sqlite3）还会阻塞事件循环，导致CPU低但响应停滞。
SQL字符串拼接在异步任务里为什么特别危险？
直接用 f"SELECT * FROM users WHERE id = {user_id}" 这类拼接方式，在异步任务中不是“慢一点”的问题，而是会立刻触发 SQL 注入——尤其当任务参数来自 HTTP 请求、队列消息或文件输入时。异步本身不改变 SQL 注入原理，但会让错误更隐蔽：一个被注入的恶意查询可能混在几十个并发协程里执行，日志里只显示 sqlite3.OperationalError 或空结果，根本看不出是攻击。
更麻烦的是，某些 ORM（比如未配置为异步模式的 SQLAlchemy Core）在 await 上下文中调用同步执行函数，会导致事件循环被阻塞，整个 asyncio 任务队列卡住，表现为 CPU 占用低但响应全无。
- 永远不用
str.format()或%拼接 SQL 中的用户输入 - 避免在
async def函数里调用sqlite3.connect().execute()等同步接口 - 确认你用的数据库驱动明确支持异步（如
aiosqlite、databases、asyncpg），而不是“只是能 import”
用 aiosqlite 构建安全异步 SQL 任务的最小可行写法
aiosqlite 是 SQLite 在 asyncio 场景下的事实标准，它封装了线程安全的连接池，并强制所有操作走 await 路径，天然规避同步阻塞。关键不是“能不能用”，而是怎么组织结构才不容易漏掉 await 或误开连接。
下面这个模式能覆盖 90% 的轻量级需求：
import aiosqlite
<p>async def fetch_user_by_id(user_id: int) -> dict | None:</p><h1>✅ 正确：每次任务都新建连接，由 aiosqlite 自动管理生命周期</h1><pre class="brush:php;toolbar:false;">async with aiosqlite.connect("app.db") as db:
async with db.execute(
"SELECT id, name, email FROM users WHERE id = ?", (user_id,)
) as cursor:
row = await cursor.fetchone()
return dict(row) if row else None
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
-
?占位符是必须的，(user_id,)必须是元组（哪怕只有一个参数） -
async with aiosqlite.connect(...)比手动await db.close()更可靠，避免连接泄露 - 不要复用
db对象跨多个async def—— 它不是线程安全的，也不是协程安全的
在 Celery 里调度异步 SQL 任务要注意什么？
Celery 默认运行在同步环境，它的 worker 进程不自带事件循环。如果你直接把 fetch_user_by_id 注册成 Celery 任务，会报错：RuntimeWarning: coroutine 'fetch_user_by_id' was never awaited，或者更糟——静默失败。
正确做法只有两种，没有中间选项：
- 用
asyncio.run()包一层（仅限简单任务，且不能嵌套调用其他 async 函数）：@app.task
def celery_fetch_user(user_id):
return asyncio.run(fetch_user_by_id(user_id)) - 改用支持原生 async 的任务框架（如
arq或selvage），或者升级 Celery 5.3+ 并启用worker_pool="gevent"+ 显式启动事件循环（复杂度陡增） - 最稳妥的折中：把异步 SQL 封装进同步函数，用
loop.run_in_executor()托管到线程池，但要小心 SQLite 的check_same_thread=False配置和连接复用冲突
为什么 ormar 或 databases 比裸写 aiosqlite 更适合生产任务？
裸用 aiosqlite 适合快速验证或脚本，但一旦任务涉及关联查询、事务、模型校验或批量写入，手写 SQL 的维护成本会指数上升。这时候 ormar 或 databases 提供的抽象就不是“方便”，而是“必要”。
例如，用 ormar 写带外键的插入：
class User(ormar.Model):
class Meta:
database = database
metadata = metadata
id: int = ormar.Integer(primary_key=True)
profile: Profile = ormar.ForeignKey(Profile) # 自动处理 JOIN 和约束
<h1>✅ 一行代码完成关联插入，且类型检查、SQL 注入防护、事务回滚全内置</h1><p>await User.objects.select_related("profile").create(
name="Alice", profile=await Profile.objects.get(name="default")
)
</p>
容易被忽略的一点是：这些库默认开启查询日志（log_level=logging.DEBUG），但在 Celery worker 里如果不显式配置日志处理器，所有 SQL 日志都会丢失。别等线上出问题才想起加 logging.getLogger("databases").setLevel(logging.DEBUG)。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
