Building a CRUD App with FastAPI, MongoDB, and Beanie

In this tutorial, you'll learn how to develop an asynchronous API with FastAPI and MongoDB. We'll be using the Beanie ODM library to interact with MongoDB asynchronously.

Contents

Objectives

By the end of this tutorial, you will be able to:

1. Explain what Beanie ODM is and why you may want to use it
2. Interact with MongoDB asynchronously using Beanie ODM
3. Develop a RESTful API with Python and FastAPI

Why Beanie ODM?

Beanie is an asynchronous object-document mapper (ODM) for MongoDB, which supports data and schema migrations out-of-the-box. It uses Motor, as an asynchronous database engine, and Pydantic.

While you could simply use Motor, Beanie provides an additional abstraction layer, making it much easier to interact with collections inside a Mongo database.

Want to just use Motor? Check out Building a CRUD App with FastAPI and MongoDB.

Initial Setup

Start by creating a new folder to hold your project called "fastapi-beanie":

$ mkdir fastapi-beanie
$ cd fastapi-beanie

Next, create and activate a virtual environment:

$ python3.10 -m venv venv
$ source venv/bin/activate
(venv)$ export PYTHONPATH=$PWD

Feel free to swap out venv and Pip for Poetry or Pipenv. For more, review Modern Python Environments.

Next, create the following files and folders:

├── app
│   ├── __init__.py
│   ├── main.py
│   └── server
│       ├── app.py
│       ├── database.py
│       ├── models
│       └── routes
└── requirements.txt

Add the following dependencies to your requirements.txt file:

beanie==1.11.0
fastapi==0.78.0
uvicorn==0.17.6

Install the dependencies from your terminal:

(venv)$ pip install -r requirements.txt

In the app/main.py file, define an entry point for running the application:

import uvicorn

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

Here, we instructed the file to run a Uvicorn server on port 8000 and reload on every file change.

Before starting the server via the entry point file, create a base route in app/server/app.py:

from fastapi import FastAPI

app = FastAPI()


@app.get("/", tags=["Root"])
async def read_root() -> dict:
    return {"message": "Welcome to your beanie powered app!"}

Run the entry point file from your console:

(venv)$ python app/main.py

Navigate to http://localhost:8000 in your browser. You should see:

{
  "message": "Welcome to your beanie powered app!"
}

What Are We Building?

We'll be building a product review application that allow us perform the following operations:

• Create reviews
• Read reviews
• Update reviews
• Delete reviews

Before diving into writing the routes, let's use Beanie to configure the database model for our application.

Database Schema

Beanie allows you to create documents that can then be used to interact with collections in the database. Documents represent your database schema. They can be defined by creating child classes that inherit the Document class from Beanie. The Document class is powered by Pydantic's BaseModel, which makes it easy to define collections and database schema as well as example data displayed in the interactive Swagger docs page.

Example:

from beanie import Document


class TestDrivenArticle(Document):
    title: str
    content: str
    date: datetime
    author: str

The document defined represents how articles will be stored in the database. However, it's a normal document class with no database collection associated with it. To associate a collection, you simple need to add a Settings class as a subclass:

from beanie import Document


class TestDrivenArticle(Document):
    title: str
    content: str
    date: datetime
    author: str


    class Settings:
        name = "testdriven_collection"

Now that we have an idea of how schemas are created, we'll create the schema for our application. In the "app/server/models" folder, create a new file called product_review.py:

from datetime import datetime

from beanie import Document
from pydantic import BaseModel
from typing import Optional


class ProductReview(Document):
    name: str
    product: str
    rating: float
    review: str
    date: datetime = datetime.now()

    class Settings:
        name = "product_review"

Since the Document class is powered by Pydantic, we can define example schema data to make it easier for developers to use the API from the interactive Swagger docs.

Add the Config subclass like so:

from datetime import datetime

from beanie import Document
from pydantic import BaseModel
from typing import Optional


class ProductReview(Document):
    name: str
    product: str
    rating: float
    review: str
    date: datetime = datetime.now()

    class Settings:
        name = "product_review"

    class Config:
        schema_extra = {
            "example": {
                "name": "Abdulazeez",
                "product": "TestDriven TDD Course",
                "rating": 4.9,
                "review": "Excellent course!",
                "date": datetime.now()
            }
        }

So, in the code block above, we defined a Beanie document called ProductReview that represents how a product review will be stored. We also defined the collection, product_review, where the data will be stored.

We'll use this schema in the route to enforce the proper request body.

Lastly, let's define the schema for updating a product review:

class UpdateProductReview(BaseModel):
    name: Optional[str]
    product: Optional[str]
    rating: Optional[float]
    review: Optional[str]
    date: Optional[datetime]

    class Config:
        schema_extra = {
            "example": {
                "name": "Abdulazeez Abdulazeez",
                "product": "TestDriven TDD Course",
                "rating": 5.0,
                "review": "Excellent course!",
                "date": datetime.now()
            }
        }

The UpdateProductReview class above is of type BaseModel, which allows us to make changes to only the fields present in the request body.

With the schema in place, let's set up MongoDB and our database before proceeding to write the routes.

MongoDB

In this section, we'll wire up MongoDB and configure our application to communicate with it.

According to Wikipedia, MongoDB is a cross-platform document-oriented database program. Classified as a NoSQL database program, MongoDB uses JSON-like documents with optional schemas.

MongoDB Setup

If you don't have MongoDB installed on your machine, refer to the Installation guide from the docs. Once installed, continue with the guide to run the mongod daemon process. Once done, you can verify that MongoDB is up and running, by connecting to the instance via the mongo shell command:

$ mongo

For reference, this tutorial uses MongoDB Community Edition v5.0.7.

$ mongo --version
MongoDB shell version v5.0.7

Build Info: {
    "version": "5.0.7",
    "gitVersion": "b977129dc70eed766cbee7e412d901ee213acbda",
    "modules": [],
    "allocator": "system",
    "environment": {
        "distarch": "x86_64",
        "target_arch": "x86_64"
    }
}

Setting up the Database

In database.py, add the following:

from beanie import init_beanie
import motor.motor_asyncio

from app.server.models.product_review import ProductReview


async def init_db():
    client = motor.motor_asyncio.AsyncIOMotorClient(
        "mongodb://localhost:27017/productreviews"
    )

    await init_beanie(database=client.db_name, document_models=[ProductReview])

In the code block above, we imported the init_beanie method which is responsible for initializing the database engine powered by motor.motor_asyncio. The init_beanie method takes two arguments:

1. database - The name of the database to be used.
2. document_models - A list of document models defined -- the ProductReview model, in our case.

The init_db function will be invoked in the application startup event. Update app.py to include the startup event:

from fastapi import FastAPI

from app.server.database import init_db


app = FastAPI()


@app.on_event("startup")
async def start_db():
    await init_db()


@app.get("/", tags=["Root"])
async def read_root() -> dict:
    return {"message": "Welcome to your beanie powered app!"}

Now that we have our database configurations in place, let's write the routes.

Routes

In this section, we'll build the routes to perform CRUD operations on your database from the application:

1. POST review
2. GET single review and GET all reviews
3. PUT single review
4. DELETE single review

In the "routes" folder, create a file called product_review.py:

from beanie import PydanticObjectId
from fastapi import APIRouter, HTTPException
from typing import List

from app.server.models.product_review import ProductReview, UpdateProductReview


router = APIRouter()

In the code block above, we imported PydanticObjectId, which will be used for type hinting the ID argument when retrieving a single request. We also imported the APIRouter class that's responsible for handling route operations. We also imported the model class that we defined earlier.

Beanie document models allow us interact with the database directly with less code. For example, to retrieve all records in a database collection, all we have to do is:

data = await ProductReview.find_all().to_list()
return data # A list of all records in the collection.

Before we proceed to writing the route function for the CRUD operations, let's register the route in app.py:

from fastapi import FastAPI

from app.server.database import init_db
from app.server.routes.product_review import router as Router


app = FastAPI()
app.include_router(Router, tags=["Product Reviews"], prefix="/reviews")


@app.on_event("startup")
async def start_db():
    await init_db()


@app.get("/", tags=["Root"])
async def read_root() -> dict:
    return {"message": "Welcome to your beanie powered app!"}

Create

In routes/product_review.py, add the following:

@router.post("/", response_description="Review added to the database")
async def add_product_review(review: ProductReview) -> dict:
    await review.create()
    return {"message": "Review added successfully"}

Here, we defined the route function, which takes an argument of the type ProductReview. As stated earlier, the document class can interact with the database directly.

The new record is created by calling the create() method.

The route above expects a similar payload as this:

{
  "name": "Abdulazeez",
  "product": "TestDriven TDD Course",
  "rating": 4.9,
  "review": "Excellent course!",
  "date": "2022-05-17T13:53:17.196135"
}

Test the route:

$ curl -X 'POST' \
  'http://0.0.0.0:8000/reviews/' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "Abdulazeez",
  "product": "TestDriven TDD Course",
  "rating": 4.9,
  "review": "Excellent course!",
  "date": "2022-05-17T13:53:17.196135"
}'

The request above should return a successful message:

{
  "message": "Review added successfully"
}

Read

Next up are the routes that enables us to retrieve a single review and all reviews present in the database:

@router.get("/{id}", response_description="Review record retrieved")
async def get_review_record(id: PydanticObjectId) -> ProductReview:
    review = await ProductReview.get(id)
    return review


@router.get("/", response_description="Review records retrieved")
async def get_reviews() -> List[ProductReview]:
    reviews = await ProductReview.find_all().to_list()
    return reviews

In the code block above, we defined two functions:

1. In the first function, the function takes an ID of type ObjectiD, the default encoding for MongoDB IDs. The record is retrieved using the get() method.
2. In the second, we retrieved all the reviews using the find_all() method. The to_list() method is appended so the results are returned in a list.

Another method that can be used to retrieve a single entry is the find_one() method which takes a condition. For example:

# Return a record who has a rating of 4.0
await ProductReview.find_one(ProductReview.rating == 4.0)

Let's test the first route to retrieve all records:

$ curl -X 'GET' \
  'http://0.0.0.0:8000/reviews/' \
  -H 'accept: application/json'

Response:

[
  {
    "_id": "62839ad1d9a88a040663a734",
    "name": "Abdulazeez",
    "product": "TestDriven TDD Course",
    "rating": 4.9,
    "review": "Excellent course!",
    "date": "2022-05-17T13:53:17.196000"
  }
]

Next, let's test the route for retrieving a single record matching a supplied ID:

$ curl -X 'GET' \
  'http://0.0.0.0:8000/reviews/62839ad1d9a88a040663a734' \
  -H 'accept: application/json'

Response:

{
  "_id": "62839ad1d9a88a040663a734",
  "name": "Abdulazeez",
  "product": "TestDriven TDD Course",
  "rating": 4.9,
  "review": "Excellent course!",
  "date": "2022-05-17T13:53:17.196000"
}

Update

Next, let's write the route for updating the review record:

@router.put("/{id}", response_description="Review record updated")
async def update_student_data(id: PydanticObjectId, req: UpdateProductReview) -> ProductReview:
    req = {k: v for k, v in req.dict().items() if v is not None}
    update_query = {"$set": {
        field: value for field, value in req.items()
    }}

    review = await ProductReview.get(id)
    if not review:
        raise HTTPException(
            status_code=404,
            detail="Review record not found!"
        )

    await review.update(update_query)
    return review

In this function, we filtered out fields that aren't updated to prevent overwriting existing fields with None.

To update a record, an update query is required. We defined an update query that overwrites the existing fields with the data passed in the request body. We then checked if the record exists. It it exist, it gets updated and the updated record is returned, otherwise a 404 exception is raised.

Let's test the route:

$ curl -X 'PUT' \
  'http://0.0.0.0:8000/reviews/62839ad1d9a88a040663a734' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "Abdulazeez Abdulazeez",
  "product": "TestDriven TDD Course",
  "rating": 5
}'

Response:

{
  "_id": "62839ad1d9a88a040663a734",
  "name": "Abdulazeez Abdulazeez",
  "product": "TestDriven TDD Course",
  "rating": 5.0,
  "review": "Excellent course!",
  "date": "2022-05-17T13:53:17.196000"
}

Delete

Lastly, let's write the route responsible for deleting a record:

@router.delete("/{id}", response_description="Review record deleted from the database")
async def delete_student_data(id: PydanticObjectId) -> dict:
    record = await ProductReview.get(id)

    if not record:
        raise HTTPException(
            status_code=404,
            detail="Review record not found!"
        )

    await record.delete()
    return {
        "message": "Record deleted successfully"
    }

So, we first checked if the record exists before proceeding to delete the record. The record is deleted by calling the delete() method.

Let's test the route:

$ curl -X 'DELETE' \
  'http://0.0.0.0:8000/reviews/62839ad1d9a88a040663a734' \
  -H 'accept: application/json'

Response:

{
  "message": "Record deleted successfully"
}

We have successfully built a CRUD app powered by FastAPI, MongoDB, and Beanie ODM.

Conclusion

In this tutorial, you learned how to create a CRUD application with FastAPI, MongoDB, and Beanie ODM. Perform a quick self-check by reviewing the objectives at the beginning of the tutorial, you can find the code used in this tutorial on GitHub.

Looking for more?

1. Set up unit and integration tests with pytest.
2. Add additional routes.
3. Create a GitHub repo for your application and configure CI/CD with GitHub Actions.

Check out the Test-Driven Development with FastAPI and Docker course to learn more about testing and setting up CI/CD for a FastAPI app.

Cheers!