FastAPI is a web framework for building APIs with Python based on standard Python type hints. It uses Starlette for the web parts and Pydantic for data validation. Some key features of FastAPI include:
NoSQL (Not only SQL) databases are non - relational databases that offer a flexible data model. There are several types of NoSQL databases:
First, install the necessary libraries:
pip install fastapi uvicorn pymongo
Here is a simple example of connecting FastAPI to MongoDB:
from fastapi import FastAPI
from pymongo import MongoClient
app = FastAPI()
# Connect to MongoDB
client = MongoClient("mongodb://localhost:27017/")
db = client["test_database"]
collection = db["test_collection"]
@app.get("/")
async def root():
return {"message": "Hello, World!"}
In this example, we first import the required libraries. Then we create a FastAPI
application instance. We connect to a local MongoDB instance, select a database, and a collection.
@app.post("/items/")
async def create_item(item: dict):
result = collection.insert_one(item)
return {"inserted_id": str(result.inserted_id)}
@app.get("/items/{item_id}")
async def read_item(item_id: str):
item = collection.find_one({"_id": ObjectId(item_id)})
if item:
item["_id"] = str(item["_id"])
return item
return {"message": "Item not found"}
@app.put("/items/{item_id}")
async def update_item(item_id: str, updated_item: dict):
result = collection.update_one({"_id": ObjectId(item_id)}, {"$set": updated_item})
if result.modified_count > 0:
return {"message": "Item updated successfully"}
return {"message": "Item not found or not updated"}
@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
result = collection.delete_one({"_id": ObjectId(item_id)})
if result.deleted_count > 0:
return {"message": "Item deleted successfully"}
return {"message": "Item not found or not deleted"}
When working with NoSQL databases in FastAPI, it’s important to handle errors properly. For example, if the database connection fails or a query returns an unexpected result, the API should return an appropriate error message.
from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: str):
try:
item = collection.find_one({"_id": ObjectId(item_id)})
if item:
item["_id"] = str(item["_id"])
return item
raise HTTPException(status_code=404, detail="Item not found")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# database.py
from pymongo import MongoClient
client = MongoClient("mongodb://localhost:27017/")
db = client["test_database"]
collection = db["test_collection"]
def get_item(item_id):
return collection.find_one({"_id": ObjectId(item_id)})
# main.py
from fastapi import FastAPI
from database import get_item
from bson.objectid import ObjectId
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: str):
item = get_item(item_id)
if item:
item["_id"] = str(item["_id"])
return item
return {"message": "Item not found"}
Combining FastAPI with NoSQL databases provides a powerful solution for building modern, high - performance APIs. FastAPI’s speed and automatic documentation, along with the flexibility of NoSQL databases, make it an ideal choice for many applications. By understanding the fundamental concepts, usage methods, common practices, and best practices, developers can efficiently build and maintain robust APIs.