FastAPI 依赖注入:构建高可维护性与可测试性 API

各位观众,各位朋友,各位屏幕前的码农们!欢迎来到“FastAPI 依赖注入:构建高可维护性与可测试性 API”讲座现场。今天,咱们要聊聊 FastAPI 框架中一个超级给力的特性——依赖注入。这玩意儿听起来有点高大上,但其实啊,它就像咱们生活中的外卖小哥,专门负责给你送餐(依赖),让你省心省力,专注于享用美食(核心业务逻辑)。

什么是依赖注入? 别怕,咱先聊点轻松的

在编程世界里,依赖指的是一个对象需要另一个对象来完成自己的工作。 比如说,咱们有个 UserService 类,它需要 Database 类来存储用户信息。 那么,UserService 就依赖于 Database

传统的做法,通常是 UserService 自己去创建 Database 实例:

class Database:
    def __init__(self):
        self.connection = "数据库连接"  # 模拟数据库连接

class UserService:
    def __init__(self):
        self.db = Database()  # UserService 自己创建 Database 实例

    def create_user(self, username: str):
        # 使用 self.db 操作数据库
        print(f"用户 {username} 已创建,使用 {self.db.connection}")

这样做的问题是啥?耦合度太高! UserServiceDatabase 紧紧地绑在一起,就像一对热恋的情侣,你想拆散他们?难!

如果有一天,你想换个数据库,或者想在测试的时候用一个假的 Database (比如一个内存数据库),那就麻烦了,你得改 UserService 的代码! 这就像你想换个外卖平台,结果发现你只能通过原来的APP点餐,太坑爹了!

依赖注入就是来解决这个问题的。它就像一个媒婆,负责把 Database “注入” 到 UserService 中,而不是让 UserService 自己去找 Database。这样,UserService 就不知道 Database 是怎么来的,它只管用就行了。

FastAPI 的依赖注入: 外卖小哥闪亮登场

FastAPI 内置了非常强大的依赖注入系统,它使用 Python 的类型提示来实现依赖关系的声明和解析。这使得代码更加简洁、易读,并且易于测试。

1. 定义依赖

首先,咱们定义一个依赖,也就是咱们的外卖小哥。这个外卖小哥负责提供 Database 实例:

class Database:
    def __init__(self):
        self.connection = "数据库连接"  # 模拟数据库连接

    def get_connection(self):
      return self.connection

async def get_db():
    db = Database()
    try:
        yield db
    finally:
        # 这里可以进行数据库连接的清理工作,比如关闭连接
        print("数据库连接已关闭")

注意几个关键点:

  • get_db 是一个函数,它返回一个 Database 实例。
  • yield db 使用了 Python 的 yield 关键字,这使得 get_db 成为一个生成器函数。 这样可以确保在使用完 Database 实例后,能够执行一些清理工作(比如关闭数据库连接)。这非常重要!
  • try...finally 块确保了清理工作无论是否发生异常都会被执行。

2. 在 FastAPI 路由中使用依赖

现在,咱们可以在 FastAPI 路由中使用这个依赖了:

from fastapi import FastAPI, Depends

app = FastAPI()

class Database:
    def __init__(self):
        self.connection = "数据库连接"  # 模拟数据库连接

    def get_connection(self):
      return self.connection

async def get_db():
    db = Database()
    try:
        yield db
    finally:
        # 这里可以进行数据库连接的清理工作,比如关闭连接
        print("数据库连接已关闭")

@app.get("/users/")
async def list_users(db: Database = Depends(get_db)):
    # 使用 db 查询用户信息
    connection = db.get_connection()
    return {"message": f"用户列表,使用 {connection} 查询"}

看看发生了什么?

  • db: Database = Depends(get_db) 这行代码告诉 FastAPI: “嘿,这个 db 参数需要一个 Database 实例,而且这个实例要从 get_db 函数那里获取。”
  • FastAPI 会自动调用 get_db 函数,获取 Database 实例,并将其传递给 list_users 函数。 整个过程就像外卖小哥把餐送到你手里一样,你只需要安心享用美食就行了。

3. 依赖之间的依赖: 外卖小哥的接力赛

依赖注入还可以嵌套,也就是说,一个依赖可以依赖于另一个依赖。 这就像外卖小哥的接力赛,一个外卖小哥把餐送到另一个外卖小哥手里,然后另一个外卖小哥再把餐送到你手里。

例如,咱们可以创建一个 UserService 依赖,它依赖于 Database 依赖:

class UserService:
    def __init__(self, db: Database):
        self.db = db

    def get_user(self, user_id: int):
        # 使用 self.db 查询用户信息
        connection = self.db.get_connection()
        return {"user_id": user_id, "message": f"用户信息,使用 {connection} 查询"}

async def get_user_service(db: Database = Depends(get_db)):
    return UserService(db)

@app.get("/users/{user_id}")
async def get_user(user_id: int, user_service: UserService = Depends(get_user_service)):
    return user_service.get_user(user_id)
  • get_user_service 依赖于 get_db,它接收 Database 实例,并创建一个 UserService 实例。
  • get_user 依赖于 get_user_service,它接收 UserService 实例,并调用其 get_user 方法。

依赖注入的好处: 让你笑出声

依赖注入带来了很多好处,让你在开发过程中笑出声:

  • 高可测试性: 你可以轻松地替换依赖,以便进行单元测试。 比如,你可以用一个假的 Database 实例来测试 UserService,而不需要连接到真实的数据库。 这就像你可以用一个假的菜品来测试外卖小哥的服务质量,而不需要真的点一份外卖。

    import pytest
    from fastapi.testclient import TestClient
    
    # 假设 FastAPI app 已经在别的文件中定义
    from main import app, get_db, Database, UserService, get_user_service  # 替换为你的实际文件
    
    # 创建一个假的 Database 类
    class FakeDatabase:
        def __init__(self):
            self.connection = "假的数据库连接"
    
        def get_connection(self):
            return self.connection
    
    # 重写 get_db 依赖,使其返回 FakeDatabase 实例
    async def override_get_db():
        db = FakeDatabase()
        try:
            yield db
        finally:
            pass # 不需要清理
    
    # 将重写的依赖添加到 FastAPI app 中
    app.dependency_overrides[get_db] = override_get_db
    
    # 创建一个测试客户端
    client = TestClient(app)
    
    # 编写测试用例
    def test_list_users():
        response = client.get("/users/")
        assert response.status_code == 200
        assert response.json() == {"message": "用户列表,使用 假的数据库连接 查询"}
    
    def test_get_user():
        response = client.get("/users/1")
        assert response.status_code == 200
        assert response.json() == {"user_id": 1, "message": "用户信息,使用 假的数据库连接 查询"}
    
    # 运行测试用例 (使用 pytest 命令)
    # pytest your_test_file.py

    在这个例子中,我们定义了一个 FakeDatabase 类,它模拟了 Database 类的行为,但实际上并没有连接到真实的数据库。然后,我们使用 app.dependency_overrides 来重写 get_db 依赖,使其返回 FakeDatabase 实例。这样,在测试 list_usersget_user 路由时,就会使用 FakeDatabase,而不是真实的 Database

  • 高可维护性: 代码更加模块化,易于理解和修改。 你可以轻松地替换依赖,而不需要修改核心业务逻辑。 这就像你可以轻松地更换外卖平台,而不需要重新装修你的厨房。
  • 松耦合: 类之间的依赖关系更加松散,降低了代码的耦合度。 这就像你和外卖小哥之间的关系,你只需要知道他能送餐就行了,不需要知道他住在哪里,开什么车。
  • 代码重用: 你可以将依赖注入到多个地方,提高代码的重用性。 这就像一个外卖小哥可以同时给多个人送餐一样。

高级用法: 让你的代码更上一层楼

FastAPI 的依赖注入系统还支持一些高级用法,让你的代码更上一层楼:

  • 类作为依赖: 你可以直接使用类作为依赖。 FastAPI 会自动创建类的实例,并将其注入到需要的地方。

    class AuthService:
        def __init__(self):
            self.secret_key = "超级密钥"
    
        def authenticate(self, token: str):
            if token == self.secret_key:
                return True
            return False
    
    @app.get("/protected/")
    async def protected_route(auth_service: AuthService = Depends(AuthService)):
        if auth_service.authenticate("超级密钥"):
            return {"message": "访问受保护的资源"}
        return {"message": "未授权"}
  • use_cache=True 你可以使用 use_cache=True 来缓存依赖的结果。 这意味着,如果多个路由需要同一个依赖,那么 FastAPI 只会调用一次依赖函数,并将结果缓存起来,供后续使用。 这就像外卖小哥只送一次餐,然后把餐分给多个人一样。 (注意,只适用于单次请求的生命周期内)

    async def expensive_operation():
        # 模拟一个耗时的操作
        import time
        time.sleep(2)
        return "操作结果"
    
    @app.get("/route1/")
    async def route1(result: str = Depends(expensive_operation, use_cache=True)):
        return {"message": f"Route 1: {result}"}
    
    @app.get("/route2/")
    async def route2(result: str = Depends(expensive_operation, use_cache=True)):
        return {"message": f"Route 2: {result}"}

    在这个例子中,expensive_operation 函数只会被调用一次,route1route2 都会使用缓存的结果。

  • Path, Query, Header, Cookie, Body, Form, File 你可以使用这些类来声明依赖,并从请求的路径、查询参数、头部、Cookie、请求体、表单、文件等地方获取数据。 这就像你可以指定外卖小哥从哪个餐厅取餐,送到哪个地址一样。

    from fastapi import Path, Query, Header
    
    @app.get("/items/{item_id}")
    async def read_item(
        item_id: int = Path(..., title="The ID of the item to get"),
        q: str = Query(None, alias="item-query"),
        user_agent: str = Header(None),
    ):
        results = {"item_id": item_id}
        if q:
            results.update({"q": q})
        if user_agent:
            results.update({"user_agent": user_agent})
        return results

最佳实践: 让你的代码更优雅

以下是一些使用 FastAPI 依赖注入的最佳实践:

  • 保持依赖函数的简洁: 依赖函数应该只负责创建和返回依赖,不要在里面编写复杂的业务逻辑。
  • 使用类型提示: 使用类型提示可以提高代码的可读性和可维护性,并且可以让 FastAPI 更好地进行依赖解析。
  • 合理使用 use_cache=True 只在需要缓存结果的时候才使用 use_cache=True,避免不必要的缓存。
  • 使用依赖覆盖进行测试: 使用 app.dependency_overrides 来重写依赖,以便进行单元测试。
  • 将依赖函数放在单独的模块中: 将依赖函数放在单独的模块中,可以提高代码的模块化程度。

总结: 让你的 API 飞起来

FastAPI 的依赖注入系统是一个非常强大的工具,它可以帮助你构建高可维护性、高可测试性的 API。 掌握了依赖注入,你就掌握了构建优秀 API 的关键。 就像掌握了外卖平台的正确使用方法,你就可以足不出户,享用各种美食。

希望今天的讲座对大家有所帮助。 记住,编程的世界充满了乐趣,只要你愿意学习,就一定能成为一名优秀的程序员! 感谢大家的收看,我们下期再见!

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注