sachinruk
sachinruk

Reputation: 9869

Running fastapi within class

import uvicorn
from fastapi import FastAPI

# 2. Create the app object
app = FastAPI()

# 3. Index route, opens automatically on http://127.0.0.1:8000
class RunModel():
    @app.get('/')
    def index(self):
        return {'message': 'Hello'}

    @app.get('/predict')
    def get_res(self, feat1: float, feat2:float):
        res = feat1 + feat2
        return {'result': f'{res:.4f}'}

run_model = RunModel()
# 5. Run the API with uvicorn
#    Will run on http://127.0.0.1:8000
if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=8000)

When I run this first of all I get the error (in terminal and not browser) 422 Unprocessable Entity. Next thing is when I go to the http://localhost:8000/docs it seems like it is expecting me to enter 3 values for /predict route, the two features as expected and self. So the question is how can I use this class structure and still use fastapi (i.e. ignore self).

Upvotes: 1

Views: 8215

Answers (3)

Corbie
Corbie

Reputation: 1066

Unfortunately, all the packages wrapping FastAPI in classes, like fastapi-utils and fastapi-class seem to be deprecated, so I would suggest using FastAPI directly.

The underlying problem is, that FastAPI uses decorators like @app.get(...), which don't work well in classes.

Inheritance

So based on Kostiantyn's answer, I have written a little example using FastAPI as base class:

from typing import Any

import uvicorn
from fastapi import FastAPI
from starlette.responses import HTMLResponse, JSONResponse


class App(FastAPI):
    def __init__(self, **extra: Any):
        super().__init__(**extra)

        self.add_api_route("/", self.get_root, methods=["GET"], include_in_schema=False)
        self.add_api_route("/version", self.get_version, methods=["GET"])

    @staticmethod
    async def get_root() -> HTMLResponse:
        return HTMLResponse('<meta http-equiv="Refresh" content="0; url=\'/docs\'" />')

    async def get_version(self) -> JSONResponse:
        return JSONResponse({"FastAPI version": self.version})


if __name__ == "__main__":
    url = "https://stackoverflow.com/q/65446591/5538913"
    app = App(
        title="FastAPI from class",
        description=f"Source: <a href='{url}'>Stack Overflow</a>",
    )
    uvicorn.run(app, host="127.0.0.1", port=8000)

This way the App and its routes can be configured easily, you can even define websockets with self.add_api_websocket_route etc.


Composition

Another approach would be to define a serve method in your class and implement the endpoints there:

import asyncio
from typing import Optional

import uvicorn
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, JSONResponse


class MyClass:
    def __init__(self):
        self.version = "0.0.1"

        url = "https://stackoverflow.com/q/65446591/5538913"
        self.app = FastAPI(
            title="FastAPI from class",
            description=f"Source: <a href='{url}'>Stack Overflow</a>",
        )
        self.serving_task: Optional[asyncio.Task] = None

    async def serve(self):
        app: FastAPI = self.app

        @app.get("/", include_in_schema=False)
        async def _get_root():
            """
            Redirect to /docs
            """
            return HTMLResponse('<meta http-equiv="Refresh" content="0; url=\'/docs\'" />')

        @app.get("/version")
        async def _get_version() -> JSONResponse:
            return JSONResponse({"MyClass version": self.version, "FastAPI version": app.version})

        # serve
        config = uvicorn.Config(app, host="127.0.0.1", port=8000)
        server = uvicorn.Server(config)
        await server.serve()


if __name__ == "__main__":
    instance = MyClass()
    asyncio.run(instance.serve())

Upvotes: 3

You can do it like this:

from fastapi import FastAPI
import uvicorn

class Settings:
    def __init__(self):
        self.api_version = "v1"
        self.api_name = "my_api"
        self.db = "some db"
        self.logger = "configured logger"
        self.DEBUG = True


class MyApi:
    def __init__(self, settings):
        self.settings = settings
        self._fastapi = FastAPI(
            version=self.settings.api_version,
        )
        self._fastapi.add_api_route(
            path="/",
            endpoint=self.index,
            methods=["GET"]
        )

        self._fastapi.add_api_route(
            path="/predict",
            endpoint=self.get_res,
            methods=["POST"]
        )

    async def index(self):
        if self.settings.DEBUG:
            pass
        return {"message": "Hello"}

    async def get_res(self, feat1: float, feat2: float):
        """
        You are able to access the settings
        """
        res = feat1 + feat2
        return {"result": f"{res:.4f}", "api_version": self.settings.api_version}

    def __getattr__(self, attr):
        if hasattr(self._fastapi, attr):
            return getattr(self._fastapi, attr)
        else:
            raise AttributeError(f"{attr} not exist")

    async def __call__(self, *args, **kwargs):
        return await self._fastapi(*args, **kwargs)


settings = Settings()
app = MyApi(settings)


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

Upvotes: 6

Yagiz Degirmenci
Yagiz Degirmenci

Reputation: 20598

Use class based views from fastapi-utils.

Create a router using InferringRouter, then decorate the class with cbv object. Inside the class, you can start creating your endpoints with your router object.

import uvicorn
from fastapi import FastAPI
from fastapi_utils.cbv import cbv
from fastapi_utils.inferring_router import InferringRouter


app = FastAPI()
router = InferringRouter()


@cbv(router)
class RunModel:
    @router.get("/")
    def index(self):
        return {"message": "Hello"}

    @router.get("/predict")
    def get_res(self, feat1: float, feat2: float):
        res = feat1 + feat2
        return {"result": f"{res:.4f}"}


app.include_router(router)

Upvotes: 6

Related Questions