Class Based Views

Source module: fastapi_utils.cbv

---

As you create more complex FastAPI applications, you may find yourself frequently repeating the same dependencies in multiple related endpoints.

A common question people have as they become more comfortable with FastAPI is how they can reduce the number of times they have to copy/paste the same dependency into related routes.

fastapi_utils provides a “class-based view” decorator (@cbv) to help reduce the amount of boilerplate necessary when developing related routes.

A basic CRUD app

Consider a basic create-read-update-delete (CRUD) app where users can create “Item” instances, but only the user that created an item is allowed to view or modify it:

from typing import NewType, Optional
from uuid import UUID

import sqlalchemy as sa
from fastapi import Depends, FastAPI, Header, HTTPException
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import Session
from starlette.status import HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND

from fastapi_utils.api_model import APIMessage, APIModel
from fastapi_utils.guid_type import GUID

# Begin setup
UserID = NewType("UserID", UUID)
ItemID = NewType("ItemID", UUID)

Base = declarative_base()


class ItemORM(Base):
    __tablename__ = "item"

    item_id = sa.Column(GUID, primary_key=True)
    owner = sa.Column(GUID, nullable=False)
    name = sa.Column(sa.String, nullable=False)


class ItemCreate(APIModel):
    name: str


class ItemInDB(ItemCreate):
    item_id: ItemID
    owner: UserID


def get_jwt_user(authorization: str = Header(...)) -> UserID:
    """Pretend this function gets a UserID from a JWT in the auth header"""


def get_db() -> Session:
    """Pretend this function returns a SQLAlchemy ORM session"""


def get_owned_item(session: Session, owner: UserID, item_id: ItemID) -> ItemORM:
    item: Optional[ItemORM] = session.get(ItemORM, item_id)
    if item is not None and item.owner != owner:
        raise HTTPException(status_code=HTTP_403_FORBIDDEN)
    if item is None:
        raise HTTPException(status_code=HTTP_404_NOT_FOUND)
    return item


# End setup
app = FastAPI()


@app.post("/item", response_model=ItemInDB)
def create_item(
    *,
    session: Session = Depends(get_db),
    user_id: UserID = Depends(get_jwt_user),
    item: ItemCreate,
) -> ItemInDB:
    item_orm = ItemORM(name=item.name, owner=user_id)
    session.add(item_orm)
    session.commit()
    return ItemInDB.from_orm(item_orm)


@app.get("/item/{item_id}", response_model=ItemInDB)
def read_item(
    *,
    session: Session = Depends(get_db),
    user_id: UserID = Depends(get_jwt_user),
    item_id: ItemID,
) -> ItemInDB:
    item_orm = get_owned_item(session, user_id, item_id)
    return ItemInDB.from_orm(item_orm)


@app.put("/item/{item_id}", response_model=ItemInDB)
def update_item(
    *,
    session: Session = Depends(get_db),
    user_id: UserID = Depends(get_jwt_user),
    item_id: ItemID,
    item: ItemCreate,
) -> ItemInDB:
    item_orm = get_owned_item(session, user_id, item_id)
    item_orm.name = item.name
    session.add(item_orm)
    session.commit()
    return ItemInDB.from_orm(item_orm)


@app.delete("/item/{item_id}", response_model=APIMessage)
def delete_item(
    *,
    session: Session = Depends(get_db),
    user_id: UserID = Depends(get_jwt_user),
    item_id: ItemID,
) -> APIMessage:
    item = get_owned_item(session, user_id, item_id)
    session.delete(item)
    session.commit()
    return APIMessage(detail=f"Deleted item {item_id}")

If you look at the highlighted lines above, you can see get_db and get_jwt_user repeated in each endpoint.

The @cbv decorator

By using the fastapi_utils.cbv.cbv decorator, we can consolidate the endpoint signatures and reduce the number of repeated dependencies.

To use the @cbv decorator, you need to:

1. Create an APIRouter to which you will add the endpoints
2. Create a class whose methods will be endpoints with shared depedencies, and decorate it with @cbv(router)
3. For each shared dependency, add a class attribute with a value of type Depends
4. Replace use of the original “unshared” dependencies with accesses like self.dependency

Let’s follow these steps to simplify the example above, while preserving all of the original logic:

from typing import NewType, Optional
from uuid import UUID

import sqlalchemy as sa
from fastapi import Depends, FastAPI, Header, HTTPException
from sqlalchemy.orm import Session, declarative_base
from starlette.status import HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND

from fastapi_utils.api_model import APIMessage, APIModel
from fastapi_utils.cbv import cbv
from fastapi_utils.guid_type import GUID
from fastapi_utils.inferring_router import InferringRouter

# Begin Setup
UserID = NewType("UserID", UUID)
ItemID = NewType("ItemID", UUID)

Base = declarative_base()


class ItemORM(Base):
    __tablename__ = "item"

    item_id = sa.Column(GUID, primary_key=True)
    owner = sa.Column(GUID, nullable=False)
    name = sa.Column(sa.String, nullable=False)


class ItemCreate(APIModel):
    name: str
    owner: UserID


class ItemInDB(ItemCreate):
    item_id: ItemID


def get_jwt_user(authorization: str = Header(...)) -> UserID:
    """Pretend this function gets a UserID from a JWT in the auth header"""


def get_db() -> Session:
    """Pretend this function returns a SQLAlchemy ORM session"""


def get_owned_item(session: Session, owner: UserID, item_id: ItemID) -> ItemORM:
    item: Optional[ItemORM] = session.get(ItemORM, item_id)
    if item is not None and item.owner != owner:
        raise HTTPException(status_code=HTTP_403_FORBIDDEN)
    if item is None:
        raise HTTPException(status_code=HTTP_404_NOT_FOUND)
    return item


# End Setup
app = FastAPI()
router = InferringRouter()  # Step 1: Create a router


@cbv(router)  # Step 2: Create and decorate a class to hold the endpoints
class ItemCBV:
    # Step 3: Add dependencies as class attributes
    session: Session = Depends(get_db)
    user_id: UserID = Depends(get_jwt_user)

    @router.post("/item")
    def create_item(self, item: ItemCreate) -> ItemInDB:
        # Step 4: Use `self.<dependency_name>` to access shared dependencies
        item_orm = ItemORM(name=item.name, owner=self.user_id)
        self.session.add(item_orm)
        self.session.commit()
        return ItemInDB.from_orm(item_orm)

    @router.get("/item/{item_id}")
    def read_item(self, item_id: ItemID) -> ItemInDB:
        item_orm = get_owned_item(self.session, self.user_id, item_id)
        return ItemInDB.from_orm(item_orm)

    @router.put("/item/{item_id}")
    def update_item(self, item_id: ItemID, item: ItemCreate) -> ItemInDB:
        item_orm = get_owned_item(self.session, self.user_id, item_id)
        item_orm.name = item.name
        self.session.add(item_orm)
        self.session.commit()
        return ItemInDB.from_orm(item_orm)

    @router.delete("/item/{item_id}")
    def delete_item(self, item_id: ItemID) -> APIMessage:
        item = get_owned_item(self.session, self.user_id, item_id)
        self.session.delete(item)
        self.session.commit()
        return APIMessage(detail=f"Deleted item {item_id}")


app.include_router(router)

The highlighted lines above show the results of performing each of the numbered steps.

Note how the signature of each endpoint definition now includes only the parts specific to that endpoint.

(Also note that we’ve also used the InferringRouter here to remove the need to specify a response_model in the endpoint decorators.)

Hopefully this helps you to better reuse dependencies across endpoints!

Info

While it is not demonstrated above, you can also make use of custom instance-initialization logic by defining an __init__ method on the CBV class.

Arguments to the __init__ function are injected by FastAPI in the same way they would be for normal functions.

You should not make use of any arguments to __init__ with the same name as any annotated instance attributes on the class. Those values will be set as attributes on the class instance prior to calling the __init__ function you define, so you can still safely access them inside your custom __init__ function if desired.