FastAPI依靠注入实践：工厂模式与实例复用的优化策略

河曲智叟 · 2025-4-6 00:52:56

title: FastAPI依靠注入实践：工厂模式与实例复用的优化策略
date: 2025/04/06 01:22:25
updated: 2025/04/06 01:22:25
author: cmdragon
excerpt:
FastAPI依靠注入体系中，类依靠的默认举动是为每个请求创建新实例，可能导致性能题目。通过工厂模式控制实例创建过程，可解耦配置和服务实例化，支持依靠层级嵌套，符合单一职责原则。使用lru_cache实现带缓存的工厂模式，优化高频调用场景性能。单例模式实现真正的单例依靠，请求级别复用策略在请求处理周期内复用实例。实际应用场景包括配置中央集成和多租户体系，动态配置加载和租户感知的依靠注入。常见报错解决方案涉及422 Validation Error和依靠项初始化失败。
categories:

后端开发
FastAPI

tags:

FastAPI
依靠注入
工厂模式
实例复用
单例模式
多租户体系
性能优化

扫描二维码关注或者微信搜一搜：编程智域前端至全栈交换与发展
探索数千个预构建的 AI 应用，开启你的下一个伟大创意
FastAPI依靠注入深度实践：类依靠的工厂模式与实例复用

一、类依靠的基本原理

在FastAPI的依靠注入体系中，类作为依靠项使用时，框架会主动创建类的实例。当我们如许定义一个路由处理函数时：

@app.get("/items/")
def read_items(service: ItemService = Depends()):
return service.get_items()

复制代码

FastAPI会为每个请求创建一个新的ItemService实例。这种默认举动在某些场景下可能产生性能题目，特殊是当依靠类必要执行初始化数据库连接、加载大文件等耗时操纵时。
二、工厂模式实现

2.1 工厂函数基础实现

通过工厂模式控制实例创建过程：

class DatabaseConfig:
def __init__(self, url: str = "sqlite:///test.db"):
self.url = url
class DatabaseService:
def __init__(self, config: DatabaseConfig):
self.connection = self.create_connection(config.url)
def create_connection(self, url):
# 模拟数据库连接
print(f"Creating new connection to {url}")
return f"Connection_{id(self)}"
def get_db_service(config: DatabaseConfig = Depends()) -> DatabaseService:
return DatabaseService(config)
@app.get("/users/")
def get_users(service: DatabaseService = Depends(get_db_service)):
return {"connection": service.connection}

复制代码

这个实现的特点：

解耦配置和服务的实例化
支持依靠层级嵌套（DatabaseConfig主动注入到工厂函数）
符合单一职责原则

2.2 带缓存的工厂模式

优化高频调用场景的性能：

from fastapi import Depends
from functools import lru_cache
class AnalysisService:
def __init__(self, config: dict):
self.model = self.load_ai_model(config["model_path"])
def load_ai_model(self, path):
print(f"Loading AI model from {path}")
return f"Model_{id(self)}"
@lru_cache(maxsize=1)
def get_analysis_service(config: dict = {"model_path": "models/v1"}) -> AnalysisService:
return AnalysisService(config)
@app.get("/predict")
def make_prediction(service: AnalysisService = Depends(get_analysis_service)):
return {"model": service.model}

复制代码

缓存机制说明：

使用lru_cache实现内存缓存
maxsize=1表示只缓存最新实例
当配置参数变革时会主动创建新实例
得当模子加载等重量级初始化场景

三、实例复用策略

3.1 单例模式实现

实现真正的单例依靠：

from contextlib import contextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
class DatabaseSingleton:
_instance = None
def __new__(cls, dsn: str):
if not cls._instance:
cls._instance = super().__new__(cls)
cls._instance.engine = create_engine(dsn)
cls._instance.Session = sessionmaker(bind=cls._instance.engine)
return cls._instance
@contextmanager
def get_db_session(dsn: str = "sqlite:///test.db"):
db = DatabaseSingleton(dsn)
session = db.Session()
try:
yield session
session.commit()
except Exception as e:
session.rollback()
raise e
finally:
session.close()
@app.get("/transactions")
def get_transactions(session=Depends(get_db_session)):
return {"status": "success"}

复制代码

3.2 请求级别复用

在请求处理周期内复用实例：

from fastapi import Request
class RequestTracker:
def __init__(self, request: Request):
self.request = request
self.start_time = time.time()
@property
def duration(self):
return time.time() - self.start_time
def get_tracker(request: Request) -> RequestTracker:
if not hasattr(request.state, "tracker"):
request.state.tracker = RequestTracker(request)
return request.state.tracker
@app.get("/status")
def get_status(tracker: RequestTracker = Depends(get_tracker)):
return {"duration": tracker.duration}

复制代码

四、实际应用场景

4.1 配置中央集成

动态配置加载示例：

from pydantic import BaseSettings
class AppSettings(BaseSettings):
env: str = "dev"
api_version: str = "v1"
class Config:
env_file = ".env"
def config_factory() -> AppSettings:
return AppSettings()
def get_http_client(settings: AppSettings = Depends(config_factory)):
timeout = 30 if settings.env == "prod" else 100
return httpx.Client(timeout=timeout)

复制代码

4.2 多租户体系

租户感知的依靠注入：

class TenantContext:
def __init__(self, tenant_id: str):
self.tenant_id = tenant_id
self.config = self.load_tenant_config()
def load_tenant_config(self):
# 模拟从数据库加载配置
return {
"db_url": f"sqlite:///tenant_{self.tenant_id}.db",
"theme": "dark" if self.tenant_id == "acme" else "light"
}
def tenant_factory(tenant_id: str = Header(...)) -> TenantContext:
return TenantContext(tenant_id)
@app.get("/dashboard")
def get_dashboard(ctx: TenantContext = Depends(tenant_factory)):
return {"theme": ctx.config["theme"]}

复制代码

五、课后Quiz

工厂模式在依靠注入中的主要作用是？
A) 镌汰代码量
B) 控制实例创建过程
C) 提高路由处理速度
D) 主动天生API文档
使用lru_cache装饰器缓存服务实例时，当什么情况下会创建新实例？
A) 每次请求时
B) 输入参数变革时
C) 服务类代码修改时
D) 服务器重启时
在多租户体系中，怎样实现差别租户的数据库隔离？
A) 使用差别的路由前缀
B) 基于租户ID动态天生数据库连接
C) 为每个租户创建独立应用实例
D) 使用请求头认证

（答案：1.B 2.B 3.B）
六、常见报错解决方案

错误1：422 Validation Error

征象：

{
"detail": [
{
"loc": [
"header",
"x-tenant-id"
],
"msg": "field required",
"type": "value_error.missing"
}
]
}

复制代码

原因分析：

请求缺少必要的Header参数
工厂函数参数类型声明错误
依靠项层级结构不匹配

解决方案：

检查请求是否包含全部必需的Header
验证工厂函数的参数类型声明
使用依靠关系图工具调试：
1. uvicorn main:app --reload --debug
复制代码

错误2：依靠项初始化失败

征象：

RuntimeError: Unable to initialize service - missing config

复制代码

排查步骤：

检查依靠项的参数传递链路
验证配置对象的默认值设置
在工厂函数中添加调试日志：
1. def get_service(config: AppSettings):
2. print("Current config:", config.dict())
3. return MyService(config)
复制代码

预防发起：

为全部配置参数设置合理的默认值
使用pydantic的Field验证：
1. class AppSettings(BaseSettings):
2. db_url: str = Field(..., env="DATABASE_URL")
复制代码

		自动登录	找回密码
密码			立即注册

FastAPI依靠注入实践：工厂模式与实例复用的优化策略

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块