前一篇文章里,我把SQLite接入了FastAPI。
不过很快就发现一个问题:
如果每个接口都自己写一遍:
conn = sqlite3.connect("app.db")代码很快就重复了。
更麻烦的是,数据库连接的创建、关闭、异常处理都散落在各个接口里。
这时候就得把数据库连接单独拎出来管理。
最简单的数据库函数
可以先封装一个函数:
import sqlite3
def get_db():
conn = sqlite3.connect("app.db")
try:
yield conn
finally:
conn.close()这里用了Python的yield。
它可以让FastAPI在请求结束后执行清理逻辑。
FastAPI Dependency
FastAPI自带很方便的依赖注入机制。
from fastapi import Depends
@app.get("/users")
def get_users(db = Depends(get_db)):
cursor = db.cursor()
cursor.execute(
"SELECT id, username FROM users"
)
return cursor.fetchall()这里:
Depends(get_db)表示这个接口依赖一个数据库连接。
FastAPI会自动执行:
请求
↓
创建数据库连接
↓
执行接口
↓
返回结果
↓
关闭数据库连接数据库生命周期就不会和业务代码混在一起了。
为什么依赖注入很重要
刚开始写FastAPI的时候可能觉得:
直接创建对象不是更简单吗?
小项目确实可以。
但项目一大,就会出现:
接口A → 数据库
接口B → 数据库
接口C → 数据库
接口D → 数据库如果每个接口自己处理连接,维护起来很麻烦。
依赖注入解决的是一个很基础的问题:
公共资源该由谁创建和销毁?
数据库只是其中一个例子。
以后还可能包括:
- 当前用户
- Redis
- 配置
- API Client
- 权限验证
- 日志对象
这些都可以用类似的方式管理。
把数据库操作再封装一层
数据库连接搞定之后,还可以把SQL操作抽出来。
比如:
def find_user_by_username(db, username):
cursor = db.cursor()
cursor.execute(
"""
SELECT id, username, email
FROM users
WHERE username = ?
""",
(username,)
)
return cursor.fetchone()接口只负责业务:
@app.get("/users/{username}")
def get_user(username: str, db = Depends(get_db)):
user = find_user_by_username(db, username)
if not user:
return {"message": "User not found"}
return {
"id": user[0],
"username": user[1],
"email": user[2],
}这样职责就开始清晰了。
Router
↓
业务逻辑
↓
数据库操作
↓
SQLite别过度抽象
不过这里也要注意一个问题。
很多人学到"分层架构"之后,马上创建十几个目录:
controllers/
services/
repositories/
models/
schemas/
entities/
factories/
providers/结果一个简单的CRUD项目反而变得巨复杂。
我更倾向:
先根据项目复杂度决定抽象程度。
几十个接口的小项目,没必要为了"架构看着专业"而增加大量文件。
真正重要的是:
职责清晰
依赖明确
修改方便而不是目录越多越好。
数据库连接需要注意什么
SQLite虽然简单,但也不是完全没有限制。
比如SQLite更适合:
- 小型服务
- 低并发场景
- 本地应用
- 个人项目
以后项目需要大量并发写入,或者多个服务同时访问数据库,就得重新考虑方案。
这不代表SQLite不好。
而是:
数据库应该随着业务规模增长而升级。
总结
数据库连接管理是后端项目里很基础的一部分。
FastAPI的Dependency Injection可以让我们比较自然地解决:
资源创建
资源使用
资源释放项目继续发展之后,还要解决一个更重要的问题:
用户到底怎么登录?
这就涉及Session、Cookie和Token了。