Microservice in Python using FastAPI, SQLAlchemy

Microservices are a popular architectural pattern for building scalable and robust applications. By breaking down a monolithic application into smaller, independent services, we can achieve better modularity, maintainability, and testability.

In this blog post, we'll show you how to create a simple microservice using Python, FastAPI, SQLAlchemy, and a REST API. We'll start by setting up a database and creating the SQLAlchemy models and Pydantic schemas, then we'll use FastAPI to create a RESTful API for our microservice.

Design of Micro-service

The design of the microservice created using Python, FastAPI, SQLAlchemy, and a REST API follows the Model-View-Controlle (MVC) architectural pattern.

In the MVC pattern, the model represents the data and business logic of the application, the view is responsible for rendering the data to the user, and the template defines the structure of the view. In our microservice, the SQLAlchemy models represent the model, the Pydantic schemas define the templates, and FastAPI takes care of the view.

The SQLAlchemy models are responsible for defining the structure of the data in the database, and for performing database operations. The Pydantic schemas are used to validate the data that the API receives and to convert the data to and from JSON. Finally, FastAPI provides a simple and efficient way to define and expose the RESTful API for our microservice.

By following this design, we can easily separate the concerns of our microservice, making it easier to develop, maintain, and test. Additionally, this design allows us to reuse the Pydantic schemas and SQLAlchemy models in other parts of our application, making it easier to manage the data consistency across the entire application.

Setting up the database
First, we'll start by setting up a database for our microservice. For this example, we'll use SQLite as our database, but you can use any relational database of your choice.

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

engine = create_engine("sqlite:///./test.db")
Base.metadata.create_all(bind=engine)

In this code, we define a User model using SQLAlchemy's declarative base. The model has three columns: id, name, and email. We also create an SQLAlchemy engine for connecting to the database, and use it to create the users table in the database.

Defining the Pydantic schemas
Next, we'll define the Pydantic schemas for our API. Pydantic is a library for validating and converting Python data structures to and from JSON.

from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str
    email: str

class User(UserCreate):
    id: int

In this code, we define two Pydantic models: UserCreate and User. The UserCreate model represents the data that the API will accept when creating a new user, and the User model represents the data that the API will return when reading a user. The User model inherits from the UserCreate model, and adds an id field.

Creating the FastAPI application
Now that we have the database and schemas set up, we can start creating the FastAPI application. FastAPI is a high-performance, easy-to-use framework for building APIs with Python.

from fastapi import FastAPI
from fastapi import Depends
from fastapi import HTTPException
from sqlalchemy.orm import Session
from . import models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = models.User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/{user_id}")
async def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

In this code, we create two API endpoints using FastAPI's routing functions: create_user and read_user. The create_user endpoint allows us to create a new user by sending a POST request to the API with a JSON payload containing the user's name and email. The read_user endpoint allows us to read a user's information by sending a GET request to the API with the user's id.

Both endpoints use an SQLAlchemy session to interact with the database. We use FastAPI's Depends function to handle the database session, so that a new session is created for each API request and automatically closed after the request is handled.

Finally, we use the HTTPException class from FastAPI to return a custom error message and HTTP status code if the user is not found in the database

Running the application
To run the application, we simply need to run the following command:

uvicorn main:app --reload

This will start the FastAPI application and make it available at http://localhost:8000/. You can now start sending requests to the API to create and read users.

Conclusion
In this blog post, we showed you how to create a simple microservice using Python, FastAPI, SQLAlchemy, and a REST API. We covered the basics of setting up a database, defining Pydantic schemas, and creating a FastAPI application with API endpoints for creating and reading users. With this foundation, you can now start building your own microservices with Python and FastAPI.