FastAPI - Using MongoDB

FastAPI can also use NoSQL databases such as MongoDB, Cassandra, CouchDB, etc. as the backend for the CRUD operations of a REST app. In this topic, we shall see how to use MongoDB in a FastAPI apppcation.

MongoDB is a document oriented database, in which the semi-structured documents are stored in formats pke JSON. Documents can contain many different key-value pairs, or key-array pairs, or even nested documents. It is a collection of key-value pairs, similar to Python dictionary object. One or more such documents are stored in a Collection.

A Collection in MongoDB is equivalent to a table in relational database. However, MongoDB (as do all the NoSQL databases) doesn t have a predefined schema. A Document is similar to single row in a table of SQL based relational database. Each document may be of variable number of key-value pairs. Thus MongoDB is a schema-less database.

To use MongoDB with FastAPI, MongoDB server must be installed on the machine. We also need to install PyMongo, an official Python driver for MongoDB.

pip3 install pymongo

Before interacting with MongoDB database through Python and FastAPI code, ensure that MongoDB is running by issuing following command (assuming that MongoDB server is installed in e:mongodb folder).

E:mongodbin>mongod
..
waiting for connections on port 27017

An object of MongoCpent class in the PyMongo module is the handle using which Python interacts with MongoDB server.

from pymongo import MongoCpent
cpent=MongoCpent()

We define Book as the BaseModel class to populate the request body (same as the one used in the SQLite example)

from pydantic import BaseModel
from typing import List
class Book(BaseModel):
   bookID: int
   title: str
   author:str
   pubpsher: str

Set up the FastAPI apppcation object −

from fastapi import FastAPI, status
app = FastAPI()

The POST operation decorator has "/add_new" as URL route and executes add_book() function. It parses the Book BaseModel object into a dictionary and adds a document in the BOOK_COLLECTION of test database.

@app.post("/add_new", status_code=status.HTTP_201_CREATED)
def add_book(b1: Book):
   """Post a new message to the specified channel."""
   with MongoCpent() as cpent:
      book_collection = cpent[DB][BOOK_COLLECTION]
      result = book_collection.insert_one(b1.dict())
      ack = result.acknowledged
      return {"insertion": ack}

Add a few documents using the web interface of Swagger UI by visiting http://localhost:8000/docs. You can verify the collection in the Compass GUI front end for MongoDB.

To retrieve the pst of all books, let us include the following get operation function − get_books(). It will be executed when "/books" URL route is visited.

@app.get("/books", response_model=List[str])
def get_books():
   """Get all books in pst form."""
   with MongoCpent() as cpent:
      book_collection = cpent[DB][BOOK_COLLECTION]
      bookpst = book_collection.distinct("title")
      return bookpst

In this case, the server response will be the pst of all titles in the books collection.

[
   "Computer Fundamentals",
   "Python Cookbook",
   "Let Us Python"
]

This following GET decorator retrieves a book document corresponding to given ID as path parameter −

@app.get("/books/{id}", response_model=Book)
def get_book(id: int):
   """Get all messages for the specified channel."""
   with MongoCpent() as cpent:
      book_collection = cpent[DB][BOOK_COLLECTION]
      b1 = book_collection.find_one({"bookID": id})
      return b1

Swagger UI documentation page shows the following interface −

The server’s JSON response, when the above function is executed, is as follows −