python-3.x - 如何将额外数据传递给 fastapi APIRouter？

Question

这是我的第一个 fastapi 练习。我用 Flask 实现了我的旧模型服务代码，如下所示：

class HealthCheck(Resource):
    def __init__(self, **kwargs):
        super(HealthCheck, self).__init__()
        self._model  = kwargs['model']
        self._logger = kwargs['logger']

    def get(self):
        if self._model:
            return {"status" : "healthy"}, HTTPStatus.OK
        return {"status": "unavailable"}, HTTPStatus.BAD_REQUEST

    def put(self):
        raise MethodNotAllowed('PUT request not supported')

    # similarly other methods are disabled

# In a different module, say in App class
        
class App():
    def __init__(self, name, logger, config):
        self._logger = logger
        self._model  = load_model(config['model_path'])
        self._flask_app = Flask(name)
        api = Api(self._flask_app)

        # logger and model is passed to HealthCheck resource
        api.add_resource(HealthCheck, "/api/healthcheck", 
                         resource_class_kwargs={'model': self._model, 'logger': self._logger})
            

如何使用 fastapi APIRouter 实现相同的目标？我的示例 fastapi 实现如下：

class HealthResult(BaseModel):
    healthy: bool

health_router = fastapi.APIRouter()

@health_router.get("/healthcheck", response_model=HealthResult, name="heathcheck")
async def heartbeat() -> HealthResult:
    hb = HealthResult(healthy=True)
    return hb

# in the App module
class App():
    def __init__(self, name, logger, config):
        self._logger = logger
        self._model  = load_model(config['model_path'])
        self._api = fastapi.FastAPI(title=name)
        self._api.include_router(health_router, prefix="/api")

        # how do I pass model and logger to health_router to use that in heartbeat method?
            

我想避免对模型和记录器使用任何全局存储，并从那里访问 health_router。

另外，由于我的 fastapi 对象在 App 类中，在这种情况下如何使用 uvicorn 调用多个工作人员？

标签： python-3.xfastapiuvicornasgi

Answer 1

虽然我不喜欢它，但我修改了我的App课程以解决这个问题 - 仍在寻找更清洁的解决方案。

class App(metaclass=Singleton):
    _MODEL: Union[None, Model] = None
    _LOGGER: Union[None, CustomLogger] = None

    @classmethod
    def setLogger(cls, logger: CustomLogger) -> None:
        cls._LOGGER = logger

    @classmethod
    def getLogger(cls) -> CustomLogger:
        return cls._LOGGER

    @classmethod
    def setModel(cls, model: Model) -> None:
        assert model
        cls._MODEL = model

    @classmethod
    def getModel(cls) -> Model:
        return cls._MODEL

    def __init__(self, name: str, logger: CustomLogger, config: YAML) -> None:
        App.setLogger(logger)
        model: Model = load_model(config['model_path'])
        App.setModel(model)
        self._api = fastapi.FastAPI(title=name)
        self._api.include_router(health_router, prefix="/api")
        ....

class HealthResult(BaseModel):
    healthy: bool

health_router = fastapi.APIRouter()

@health_router.get("/healthcheck", response_model=HealthResult, name="heathcheck")
async def heartbeat() -> HealthResult:
    model: Model = App.getModel()
    hb: HealthResult = HealthResult(healthy=True) if model else HealthResult(healthy=False)
    return hb