亚洲在线久爱草,狠狠天天香蕉网,天天搞日日干久草,伊人亚洲日本欧美

為了賬號安全,請及時綁定郵箱和手機立即綁定

這就是為什么FastAPI目前尚不完全準備好用于生產環境的原因

摘要

在现代 web 开发中,高效地管理依赖关系对于创建可扩展和可维护的应用程序至关重要。依赖注入 (DI) 和控制反转 (IoC) 是两种解决这一需求的设计原则。本文探讨了如何在两个流行的 Python 框架:FastAPI 和 PyNest 中实现 DI 和 IoC。我们将介绍这两个框架,并深入探讨它们各自的 DI 实现方法,并提供一个全面的比较,以帮助您为下一个项目选择最佳框架。

依赖注入是FastAPI的阿喀琉斯之踵

FastAPI 介绍

FastAPI 是一个现代、快速(高性能)的基于 Python 3.7+ 构建 API 的 Web 框架,它基于标准的 Python 类型提示。FastAPI 提供了 OpenAPI 和 JSON Schema 的自动生成,使其成为开发人员快速高效地创建 API 的首选框架。FastAPI 的 DI 方法内置支持,利用 Python 的类型提示来无缝注入依赖。

FastAPI 的 DI 方法

在 FastAPI 中,依赖注入是通过在路径操作函数(路由处理程序)的函数签名中使用 Depends 关键字来处理的。这告诉 FastAPI 调用依赖函数,并将结果用作该参数的值。

示例 : FastAPI 依赖注入
    从 fastapi 导入 Depends, FastAPI  

    app = FastAPI()  

    # 依赖函数  
    async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):  
        return {"q": q, "skip": skip, "limit": limit}  

    # 使用依赖的路由  
    @app.get("/items/")  
    async def read_items(commons: dict = Depends(common_parameters)):  
        return commons

在这个例子中,common_parameters 是一个依赖函数,每当访问 /items/ 路由时,它会在 read_items 之前执行。

基于依赖注入的冗余

FastAPI 需要在每个路由的函数级别注入依赖。如果多个路由需要相同的依赖,必须分别注入到每个路由处理函数中。这可能导致代码重复且冗长,特别是在路由和共享依赖较多的应用程序中。

示例
    from fastapi import Depends, FastAPI  

    app = FastAPI()  

    # 依赖项  
    class Logger:  
        def __init__(self):  
            print("日志开始")  
            time.sleep(2)  
            print(f"日志启动于 - {self}")  

        def log(self, message):  
            print(f"记录 - {message}")  

    # 多个路由注入相同的依赖项  
    @app.get("/items/")  
    async def read_items(logger: Annotated[Logger, Depends(Logger)]):  
        logger.log("项目列表")  
        return {"message": "项目"}  

    @app.post("/items/")  
    async def create_item(logger: Annotated[Logger, Depends(Logger)]):  
        logger.log("创建项目")  
        return {"message": "项目已创建"}

我们可以看到,对于每个需要数据库的新路由,我们都需要显式地将其注入到我们的路由中。让我们稍微扩展一下,想象一下所有路由都需要注入我们共享的日志记录器、共享配置和数据库连接,最终会写出很多重复的代码。

在FastAPI中注入类依赖关系

让我们来探索一个场景,在FastAPI路由中注入单一和多个依赖项以及每次调用时对象重新初始化的问题。

注入单一依赖项 — 在这个例子中,我们将创建一个 Logger 对象,该对象将被注入到我们的 API 路由中。

    from fastapi import FastAPI, Depends  

    app = FastAPI()  

    class Logger:  
        def __init__(self):  
            print("Logger 开始")  
            time.sleep(2)  
            print(f"Logger 启动完成 - {self}")  
            self.params = {}  

        def log(self, message):  
            print(f"记录 - {message}")  

    @app.get("/")  
    def get(logger: Logger = Depends(Logger)):  
        logger.log("端点被访问")  
        return "Logger 工作正常"  

    if __name__ == "__main__":  
        import uvicorn  
        uvicorn.run(app, host="0.0.0.0", port=8000)

输出:

    日志启动  
    日志启动于 - <main.Logger object at 0x102f3ddf0>  
    记录 - 端点被访问  
    INFO: 127.0.0.1:64670 - "GET / HTTP/1.1" 200 OK  

    日志启动  
    日志启动于 - <main.Logger object at 0x102f3dbe0>  
    记录 - 端点被访问  
    INFO: 127.0.0.1:64670 - "GET / HTTP/1.1" 200 OK  

    日志启动  
    日志启动于 - <main.Logger object at 0x102f3dbb0>  
    记录 - 端点被访问  
    INFO: 127.0.0.1:64670 - "GET / HTTP/1.1" 200 OK

让我们来检查一下输出。我们可以看到,当我们尝试访问应用程序的根路由时,logger 对象被初始化并设置到了位置 “0x102f3ddf0”。然后,我们再次访问相同的路由,logger 对象再次被初始化,这次它被设置到了另一个内存位置。每次我们调用根路由时,仅仅为了初始化 logger 就要花费 2 秒的延迟。

现在,当我们有一个想要使用的服务,并且这个服务依赖于日志记录器时,会发生什么?

    from fastapi import FastAPI, Depends  
    import time  
    import random  

    class Logger:  
        def __init__(self):  
            print("Logger 开始")  
            time.sleep(2)  
            print(f"Logger 启动于 - {self}")  

        def log(self, message):  
            print(f"记录 - {message}")  

    class Service:  
        def __init__(self, logger: Logger = Depends(Logger)):  
            self.logger = logger  
            print("Service 开始")  
            time.sleep(1)  
            print(f"Service 启动于 - {self}")  

        def do(self):  
            self.logger.log("正在做某事")  
            return f"做某事, {random.random()}"  

    app = FastAPI()  

    @app.get("/")  
    def get(service: Service = Depends(Service)):  
        return f"{service.do()}"  

    if __name__ == "__main__":  
        import uvicorn  
        uvicorn.run(app, host="0.0.0.0", port=8000)

当然,你说得对!FastAPI 及其在 IOC 容器中管理依赖时缺乏支持,实际上导致每次调用此端点时,我们都会花费 3 秒的延迟,仅仅是为了初始化已经初始化的对象。这种模式是一种真正的反模式,并且可能暴露了 FastAPI 的阿喀琉斯之踵

经过数小时的努力尝试解决这些问题后,我意识到Fastapi本身无法做到这一点,需要一种新的、更全面的方法。这就是为什么我最终创建了PyNest,一个专注于依赖注入和模块化的Python元框架。

PyNest:一种模块化的依赖注入方法

PyNest 的 DI 系统旨在减少重复代码并简化开发流程,特别是对于大型应用。PyNest 提供了一个结构化的 DI 系统,使依赖项可以在控制器类级别注入一次,从而促进代码的重用并遵循 DRY 原则。这种结构意味着一旦依赖项被注入到控制器中,就可以在控制器的所有路由方法中使用,而无需进一步注入,从而简化了代码库。

PyNest — 更加全面的依赖注入方法

一个注入,多次使用

使用 PyNest 的模块化架构,你可以将依赖注入到控制器类的构造函数中,这样你就可以在整个路由中使用这个依赖,而无需在每个路由中反复注入。

示例 —

    from nest.core import Injectable, Controller, Get, Post  

    @Injectable  
    class Logger:  
        def __init__(self):  
            print("Logger 开始")  
            time.sleep(2)  
            print(f"Logger 启动完成 - {self}")  

        def log(self, message):  
            print(f"记录 - {message}")  

    @Controller("items")  
    class ItemsController:  

      # 一次注入  
      def __init__(self, logger: Logger):  
          self.logger = logger  

      # 多次使用  
      @Get("/")  
      async def read_items(self):  
          self.logger.log("项目列表")  
          return {"message": "项目"}  

      @Post("/{item}")  
      async def create_item(self, item: str):  
          self.logger.log("创建项目")  
          return {"message": f"项目创建完成 - {item}"}

这种模块化的方法使我们能够将任意数量的依赖项注入到控制器的构造函数中,并在类方法中访问这些依赖项。结果是代码更加简洁,无需重写代码,也不必使事情比必要的更复杂。

拥抱单例模式的力量

正如我们之前讨论的,FastAPI 的依赖注入机制最大的缺点是它没有使用单例模式来管理依赖。我们注意到依赖必须在每个传入请求时进行初始化。

在 PyNest 中,我们利用了底层的“injector”库,这是一个用于管理现代 Python 应用程序中依赖关系的包。injector 支持单例模式,以及多绑定。当一个类被标记为 Injectable 并注册为依赖项时,injector 会创建该类的一个实例并存储其引用。对这个可注入对象的每次调用都会通过 injector,injector 会返回任何可注入对象的单例实例。

让我们通过代码来分析这一点

首先,让我们从 PyNest 引入所有相关的导入。

     导入 logging  
    导入 os  

    从 nest.core 导入 (  
        Controller,  
        Delete,  
        Get,  
        Injectable,  
        Module,  
        Post,  
        Put,  
        PyNestFactory,  
    )  
    导入 time

然后我们将声明两个提供者,我们希望注入它们,以及包含逻辑层的主要服务。

    # 配置提供者  
    @Injectable()  
    class ConfigService:  
        def __init__(self):  
            time.sleep(2)  
            print(f"配置服务启动 - {self}")  
            self.config = os.environ  

        def get(self, key: str):  
            return self.config.get(key)  

    # 日志提供者  
    @Injectable()  
    class Logger:  
        def __init__(self, config_service: ConfigService):  
            time.sleep(2)  
            print(f"日志启动 - {self}")  
            self.config_service = config_service  
            self.log = logging.getLogger(__name__)  

    # 我们的主服务  
    @Injectable()  
    class ItemService:  
        def __init__(self, logger: Logger):  
            time.sleep(2)  
            print(f"项目服务启动 - {self}")  
            self.logger = logger  
            self.items = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]  

        def get(self):  
            self.logger.log.info("端点被访问")  
            return self.items  

        def post(self, item: dict):  
            self.items.append(item)  
            return self.items  

        def put(self, item: dict):  
            self.items.append(item)  
            return self.items  

        def delete(self, item: dict):  
            self.items.remove(item)  
            return self.items

现在,让我们创建一个控制器并将其注入到我们的服务中 —

    @Controller("items")  
    class ItemController:  
        def __init__(self, item_service: ItemService):  
            print("ItemController 正在启动 - {self}")  
            self.item_service = item_service  

        @Get("/")  
        def get(self):  
            return self.item_service.get()  

        @Post("/")  
        def post(self, item: dict):  
            return self.item_service.post(item)  

        @Put("/")  
        def put(self, item: dict):  
            return self.item_service.put(item)  

        @Delete("/")  
        def delete(self, item: dict):  
            return self.item_service.delete(item)

太好了,我们快完成了。现在让我们定义应用模块并运行应用 -

    @Module(  
        controllers=[ItemController],  
        providers=[Logger],  
    )  
    class AppModule:  
        pass  

    app = PyNestFactory.create(AppModule)  

    if __name__ == "__main__":  
        import uvicorn  

        uvicorn.run(app.http_server, host="0.0.0.0", port=8623)

输出 —

    ConfigService 启动 - <__main__.ConfigService 对象 at 0x10444d580>  
    Logger 启动 - <__main__.Logger 对象 at 0x10444daf0>  
    ItemService 启动 - <__main__.ItemService 对象 at 0x10444d190>  

    INFO:     启动服务器进程 [64770]  
    INFO:     等待应用启动。  
    INFO:     应用启动完成。  
    INFO:     Uvicorn 在 http://0.0.0.0:8623 运行 (按 CTRL+C 退出)  

    INFO:     127.0.0.1:63810 - "GET /items/ HTTP/1.1" 200 OK  
    INFO:     127.0.0.1:63824 - "PUT /items/ HTTP/1.1" 200 OK  
    INFO:     127.0.0.1:63840 - "POST /items/ HTTP/1.1" 200 OK

哇!这真的很厉害。我们可以看到,我们只需要初始化一次我们的 Injectable 对象,从那之后,容器就会管理这些对象的实例。

PyNest 依赖注入系统

PyNest Dependency Injection System

PyNest DI 明单
可注入对象
  • Injectable 对 Injectable 注入 : 可注入对象可以注入其他可注入对象,从而创建一个一致且统一的依赖关系层级。
控制器
  • Controller 注入 : 控制器能够注入可注入对象,使其能够根据需要将职责委托给服务和仓库。
依赖图
  • 无环依赖图:依赖关系必须形成一个有向无环图(DAG)。不应存在循环依赖,以保持可管理的依赖解析并防止运行时错误或无限循环。
依赖解析与管理
  • 在应用初始化时,IoC 容器解析所有依赖关系,创建尚未注册的对象实例,并管理这些实例以确保它们在任何注入位置都能被提供。
导出提供者
  • 一个模块可以导出提供者,这些提供者可以被应用程序中的其他模块使用或注入。
模块间提供者注入
  • 为了从另一个模块注入提供者,一个模块必须显式地导入包含所需提供者的模块。
实例引用和重用
  • 当应用程序调用注入的提供者时,它会引用已经初始化的实例并重复使用它。这可以防止不必要的提供者实例的创建,在需要时遵循单例模式等模式。
结论:为什么 PyNest 可能更适合用于 DI

PyNest 在依赖注入方面的做法在代码组织和可维护性方面提供了明显的优势,特别是在大型项目中,模块化和避免重复至关重要。通过允许在应用程序结构的较高层次注入依赖,PyNest 促进了更符合 DRY 原则的代码库,减少了错误的可能性,并简化了重构和测试的过程。

相比之下,虽然 FastAPI 的依赖注入系统有自己的优势,但它要求在函数级别进行注入,这可能会引入冗余和繁琐,随着 web 应用程序复杂性的增加,这可能会影响其可维护性和可扩展性。

资源
一个简单的解释 🚀

感谢您成为In Plain English社区的一员!在您离开之前:

點擊查看更多內容
TA 點贊

若覺得本文不錯,就分享一下吧!

評論

作者其他優質文章

正在加載中
  • 推薦
  • 評論
  • 收藏
  • 共同學習,寫下你的評論
感謝您的支持,我會繼續努力的~
掃碼打賞,你說多少就多少
贊賞金額會直接到老師賬戶
支付方式
打開微信掃一掃,即可進行掃碼打賞哦
今天注冊有機會得

100積分直接送

付費專欄免費學

大額優惠券免費領

立即參與 放棄機會
微信客服

購課補貼
聯系客服咨詢優惠詳情

幫助反饋 APP下載

慕課網APP
您的移動學習伙伴

公眾號

掃描二維碼
關注慕課網微信公眾號

舉報

0/150
提交
取消