FastAPI服务类注入端点遇模块导入错误，求正确依赖注入方案

第一步：修复模块导入错误 #

你遇到的ModuleNotFoundError是因为服务间导入路径不正确，且未将服务目录标记为Python包。按以下步骤修复：

1. 添加包标识文件
   在api/services和api/routers目录下创建空的__init__.py文件，将这两个目录标记为Python包。

2. 修改服务间的导入语句
   所有服务文件内部的导入需使用相对导入（因为它们属于同一个services包）：

   • image_segmentation_service.py：

     # 替换原导入语句
     from .aws_rekognition_service import AWSRekognitionService
     

   • segment_anything_service.py：

     # 替换原导入语句
     from .grounding_dino_service import GroundingDINOService
     

   • book_recognition_service.py：

     # 替换原所有服务导入语句
     from .grounding_dino_service import GroundingDINOService
     from .image_segmentation_service import ImageSegmentationService
     from .segment_anything_service import SegmentAnythingService
     from .visualization_service import VisualizationService
     

3. 修正配置文件路径
   book_recognition_service.py中读取配置的路径需适配运行环境，改用绝对路径方式读取：

   from pathlib import Path
   
   # 替换原路径读取代码
   config_path = Path(__file__).parent.parent / "config" / "ml_config.json"
   with open(config_path) as json_file:
       config = json.load(json_file)
   

第二步：重构服务类实现依赖注入 #

移除模块级别的服务实例化，改为通过构造函数注入依赖，让服务间的依赖关系更清晰：

1. 修改ImageSegmentationService

   class ImageSegmentationService:
       def __init__(self, aws_service: AWSRekognitionService) -> None:
           self.aws_rekognition_service = aws_service
       # 其余业务逻辑保持不变
   

2. 修改SegmentAnythingService

   class SegmentAnythingService:
       def __init__(self, grounding_dino_service: GroundingDINOService) -> None:
           self.grounding_dino_service = grounding_dino_service
       # 其余业务逻辑保持不变
   

3. 修改BookRecognitionService

   class BookRecognitionService:
       def __init__(
           self,
           grounding_dino: GroundingDINOService,
           image_segmentation: ImageSegmentationService,
           segment_anything: SegmentAnythingService,
           visualization: VisualizationService
       ) -> None:
           self.grounding_dino_service = grounding_dino
           self.image_segmentation_service = image_segmentation
           self.segment_anything_service = segment_anything
           self.visualization_service = visualization
           # 加载配置
           config_path = Path(__file__).parent.parent / "config" / "ml_config.json"
           with open(config_path) as json_file:
               self.config = json.load(json_file)
   
       # 注意：添加self参数
       async def recognize(self):
           # 业务逻辑实现
           pass
   

第三步：在FastAPI中配置依赖注入 #

推荐使用启动事件+依赖函数的方式管理服务实例，确保服务在应用启动时初始化一次（单例）：

1. 修改main.py

   from fastapi import FastAPI, Depends
   from motor.motor_asyncio import AsyncIOMotorClient
   from config import mongo_config
   from routers import auth, book
   # 导入所有服务类
   from services.aws_rekognition_service import AWSRekognitionService
   from services.grounding_dino_service import GroundingDINOService
   from services.image_segmentation_service import ImageSegmentationService
   from services.segment_anything_service import SegmentAnythingService
   from services.visualization_service import VisualizationService
   from services.book_recognition_service import BookRecognitionService
   
   app = FastAPI()
   
   @app.on_event("startup")
   async def startup_event():
       # 初始化MongoDB
       app.mongodb_client = AsyncIOMotorClient(mongo_config.MONGO_URI)
       app.mongodb = app.mongodb_client[mongo_config.MONGO_DB_NAME]
       
       # 初始化所有AI服务
       app.aws_rekognition_service = AWSRekognitionService()
       app.grounding_dino_service = GroundingDINOService()
       app.image_segmentation_service = ImageSegmentationService(app.aws_rekognition_service)
       app.segment_anything_service = SegmentAnythingService(app.grounding_dino_service)
       app.visualization_service = VisualizationService()
       app.book_recognition_service = BookRecognitionService(
           app.grounding_dino_service,
           app.image_segmentation_service,
           app.segment_anything_service,
           app.visualization_service
       )
   
   @app.on_event("shutdown")
   async def shutdown_event():
       app.mongodb_client.close()
       # 若服务需要释放资源（如模型），在此添加关闭逻辑
   
   # 定义依赖函数，提供BookRecognitionService实例
   def get_book_recognition_service():
       return app.book_recognition_service
   
   app.include_router(auth.router)
   app.include_router(book.router)
   

2. 修改book.py路由
   通过Depends注入服务实例：

   from fastapi import APIRouter, status, JSONResponse, Depends
   from services.book_recognition_service import BookRecognitionService
   from main import get_book_recognition_service
   
   router = APIRouter()
   
   @router.get("/recognize", status_code=status.HTTP_200_OK)
   async def recognize_books(
       book_service: BookRecognitionService = Depends(get_book_recognition_service)
   ):
       book_texts = await book_service.recognize()
       return JSONResponse(status_code=status.HTTP_200_OK, content=book_texts)
   

额外注意事项 #

• 确保服务初始化逻辑（如模型加载）是线程安全的，FastAPI默认使用多线程运行，启动事件是单线程执行的，适合初始化全局资源。
• 若需要更灵活的依赖管理，可以使用functools.lru_cache装饰依赖函数，实现单例服务的创建。

内容的提问来源于stack exchange，提问作者Stefan Jaspers