Class Based Views - FastAPI Utilities (original) (raw)

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 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.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_dband 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 APIRouter, 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

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 = APIRouter() # 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.

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.