Building a FastAPI Application with gRPC and MongoDB Integration

Modern applications often require efficient communication between microservices, high-speed data access, and seamless integration of multiple tools. This article will guide you through building a FastAPI application that uses gRPC for inter-service communication and MongoDB for data storage.

What You’ll Learn

1. Setting up FastAPI and MongoDB.
2. Defining and implementing gRPC services.
3. Connecting FastAPI with gRPC and MongoDB.
4. Testing the application.

Prerequisites

• Python 3.8+ installed.
• Basic understanding of FastAPI, gRPC, and MongoDB.
• MongoDB installed locally or hosted (e.g., MongoDB Atlas).
• protoc (Protocol Buffer Compiler) installed.

Setting Up the Project

Install Dependencies

Create a virtual environment and install the required Python packages:

pip install fastapi grpcio grpcio-tools pydantic motor uvicorn        

Project Structure

Organize your project for better maintainability:

fastapi_grpc_mongodb/
├── app/
│   ├── main.py
│   ├── grpc/
│   │   ├── user.proto
│   │   ├── user_pb2.py
│   │   ├── user_pb2_grpc.py
│   ├── database.py
│   ├── models.py
│   ├── services.py
│   ├── grpc_server.py
│   └── grpc_client.py
├── Dockerfile
├── requirements.txt
└── README.md        

Define gRPC Services

Creating user.proto

Define the gRPC service and messages in the app/grpc/user.proto file:

syntax = "proto3";

service UserService {
    rpc CreateUser (CreateUserRequest) returns (CreateUserResponse);
    rpc GetUser (GetUserRequest) returns (GetUserResponse);
}

message CreateUserRequest {
    string name = 1;
    string email = 2;
}

message CreateUserResponse {
    string id = 1;
}

message GetUserRequest {
    string id = 1;
}

message GetUserResponse {
    string id = 1;
    string name = 2;
    string email = 3;
}        

Generate gRPC Code

Run the protoc command to generate Python gRPC code:

python -m grpc_tools.protoc -I app/grpc --python_out=app/grpc --grpc_python_out=app/grpc app/grpc/user.proto        

This generates user_pb2.py and user_pb2_grpc.py files.

MongoDB Integration

Database Connection (database.py)

Use Motor, an asynchronous MongoDB driver:

from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client["fastapi_grpc_db"]        

Define Data Models (models.py)

Use Pydantic for data validation:

from pydantic import BaseModel

class User(BaseModel):
    id: str = None
    name: str
    email: str        

Implement gRPC Services

Create gRPC Server (grpc_server.py)

Implement the gRPC service logic:

from app.grpc import user_pb2, user_pb2_grpc
from app.database import db
from bson.objectid import ObjectId

class UserService(user_pb2_grpc.UserServiceServicer):
    async def CreateUser(self, request, context):
        user = {"name": request.name, "email": request.email}
        result = await db.users.insert_one(user)
        return user_pb2.CreateUserResponse(id=str(result.inserted_id))

    async def GetUser(self, request, context):
        user = await db.users.find_one({"_id": ObjectId(request.id)})
        if user:
            return user_pb2.GetUserResponse(
                id=str(user["_id"]), name=user["name"], email=user["email"]
            )
        context.abort(grpc.StatusCode.NOT_FOUND, "User not found")        

Start gRPC Server

import grpc
from concurrent import futures
from app.grpc import user_pb2_grpc
from app.grpc_server import UserService

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    user_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    server.wait_for_termination()

if __name__ == "__main__":
    serve()        

Connect FastAPI with gRPC

gRPC Client (grpc_client.py)

Set up the gRPC client in FastAPI:

import grpc
from app.grpc import user_pb2, user_pb2_grpc

channel = grpc.aio.insecure_channel("localhost:50051")
stub = user_pb2_grpc.UserServiceStub(channel)

async def create_user(name: str, email: str):
    request = user_pb2.CreateUserRequest(name=name, email=email)
    response = await stub.CreateUser(request)
    return response.id

async def get_user(user_id: str):
    request = user_pb2.GetUserRequest(id=user_id)
    response = await stub.GetUser(request)
    return {"id": response.id, "name": response.name, "email": response.email}        

FastAPI Routes (main.py)

Expose FastAPI endpoints to interact with gRPC and MongoDB:

from fastapi import FastAPI, HTTPException
from app.grpc_client import create_user, get_user

app = FastAPI()

@app.post("/users/")
async def create_user_endpoint(name: str, email: str):
    user_id = await create_user(name, email)
    return {"id": user_id}

@app.get("/users/{user_id}")
async def get_user_endpoint(user_id: str):
    try:
        user = await get_user(user_id)
        return user
    except grpc.aio._call.AioRpcError:
        raise HTTPException(status_code=404, detail="User not found")        

Running the Application

Start MongoDB

Ensure MongoDB is running locally or on a hosted service.

Run gRPC Server

python app/grpc_server.py        

Run FastAPI Application

uvicorn app.main:app --reload        

Testing the Application

Create a User

Use curl or a tool like Postman to create a user:

curl -X POST "http://127.0.0.1:8000/users/" -H "Content-Type: application/json" -d '{"name": "John Doe", "email": "john@example.com"}'        

Get a User

Retrieve the user data using their ID:

curl -X GET "http://127.0.0.1:8000/users/<user_id>"        

Dockerize the Application

Dockerfile

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]        

Run with Docker

Build and run the Docker container:

docker build -t fastapi-grpc-mongodb .
docker run -p 8000:8000 fastapi-grpc-mongodb        

Conclusion

By combining FastAPI, gRPC, and MongoDB, you can build high-performance microservices that are efficient, scalable, and easy to maintain. This setup is suitable for production-grade applications where performance and modularity are key.

Thank you for taking the time to read! Follow me for more insights and updates, and let’s continue to grow and learn together.