A Well-Architected FastAPI Boilerplate
Tags: python, react, javascript, fastapi
Published: 2024-02-22 10:01:52

To read more articles like this, [visit my blog](https://www.mdfaisal.com/blog)


As a project grows larger, the value of a well-architected and consistent project structure becomes more evident.

Today, we will create a well-architected FastAPI project that should be easily scalable.

We will also do it from scratch, so there will be no surprise for you!

After we are done, our project will have -

* A CRUD API ( **Of course!** )

* Alembic for migrations.

* SQL Alchemy for Database Operations.

* Different environment handling.

* Docker-compose file for local development.

* Dockerfile for production

* Linters.

You can find the repository here if you are looking for that.

[**GitHub - Mohammad-Faisal/fastapi-well-architected-boilerplate: A Well arthitected FastAPI…**](https://github.com/Mohammad-Faisal/fastapi-well-architected-boilerplate/tree/master)

If you want to see how to build one from scratch — Let’s get started!

## First step.

First, create the project directory and navigate to it.

```
mkdir fastapi-well-architected-boilerplate
cd fastapi-well-architected-boilerplate
```

Then, create the following directories and files:

```
mkdir -p src/{api,core,db}
touch src/__init__.py
touch src/{main}.py
```
 
The project structure should look like this:

```
.
├── src
│   ├── main.py
└── README.md
```

The `src` directory contains the main application code.

The `main.py` file is the entry point of the application.

## Create a virtual environment and install dependencies.

Create a virtual environment and install the required dependencies.

```
python3 -m venv venv
source venv/bin/activate
```

Install the two dependencies `fastapi` and `uvicorn` using the following command:

```
pip3 install fastapi uvicorn
```

* The `fastapi` package is a modern, fast (high-performance), web framework for building APIs with Python 3.7+ based on standard Python type hints.

* The `unicorn` package is a lightning-fast ASGI server implementation using uvloop and HTTP tools.

And create a `requirements.txt` file to store the dependencies.

```
pip3 freeze > requirements.txt
```

Add the following code to the src/main.py file.

```js
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}
```

Run the application using the following command:

```
uvicorn src.main:app --reload
```

Now open your browser and navigate to http://localhost:8000/docs to see the API documentation.

## Different environments and configuration settings

The application will have different environments, such as `development`, `testing`, and `production`. Each environment will have its configuration settings.

The configuration settings will be stored in a `.env` file in the project's root directory. The `.env` file will contain the following settings:

```
# .env
ENV=development
DATABASE_URL=sqlite:///./test.db
```

The `ENV` setting will be used to determine the current environment. The `DATABASE_URL` setting will be used to connect to the database. The `DATABASE_URL` setting will be different for each environment.

Let’s install the `pydantic-settings` package to load the configuration settings from the `.env` file.

```
pip3 install pydantic-settings
```

The `src/config.py` file contains the following code:

```js
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    ENV: str = "development"
    DATABASE_URL: str = "sqlite:///./test.db"

    class Config:
        env_file = ".env"
        
settings = Settings()
```

The `Settings` class contains the configuration settings. The `Config` class is used to load the settings from the `.env` file.

Now, you can print the configuration settings using the following code:

```js
from src.config import settings
    
print(settings.ENV)
print(settings.DATABASE_URL)
```

## Format the code using Black

The `black` package is a Python code formatter. It will format the code according to the Python PEP 8 style guide.

Install the `black` package using the following command:

```
pip3 install black
```

And create a `pyproject.toml` file in the root directory of the project with the following content:

```js
[tool.black]
line-length = 88
target-version = ['py37']
```

The `pyproject.toml` file contains the configuration settings for the `black` package. The `line-length` setting is used to specify the maximum line length. The default is **88**, but you can change it to any value.

The `target-version` setting is used to specify the Python version.

Now, you can format the code using the following command:

```
black src
```

This command will format the code in the `src` directory.

Now black is great for formatting the code but it can be a bit too aggressive. If you want to see what changes it would make without actually making them, you can use the `--diff` option:

```
black --diff src
```

If you want to see what changes it would make without actually making them, you can use the `---check` Option:

```
black --check src
```

Also, black focuses more on the code formatting. If you want to check the code for style and programming errors, you can use the `flake8` package.

## Database models and migrations

The application will use the `SQLModel` package to work with the database.

This is built on top of SQLAlchemy and Pydantic. It will allow us to define the database models using Python-type hints.

Install the `SQLAlchemy` package using the following command:

```
pip3 install sqlmodel
```

The `src/database.py` file contains the following code:

```js
from sqlmodel import create_engine, Session
from src.config import settings
  
SQLALCHEMY_DATABASE_URL = settings.DATABASE_URL

engine = create_engine(SQLALCHEMY_DATABASE_URL)
    
def get_session():
    with Session(engine) as session:
        yield session
```

This code creates a database engine and a session. The `get_session` function is used to get the session.

## Create our first model

Now, let’s create our first database model. We will create separate domains for each domain in the application.

Let’s create our user domain

```
touch src/api/user/{__init__.py,models.py}
```

This will create the following structure:

```    .
├── src
│   ├── api
│       ├── user
│           ├── __init__.py
│           ├── models.py
# ... the other stuff
└── README.md
```

Then, create the user model using the following code.

```js
from sqlmodel import Field, SQLModel
  
class User(SQLModel, table=True):
    id: int = Field(primary_key=True, index=True)
    username: str
    email: str
    password: str
```

The `User` class is a database model. It inherits from the `SQLModel` class.

The `id` field is the primary key. The `username`, `email`, and `password` fields are the columns in the database table.

Now we have the models, but we need to create the database tables.

## Database migrations with Alembic

The **Alembic** package is a database migration tool for SQLAlchemy. It will create the database and tables in the production environment.

Install the Alembic package using the following command:

```
pip3 install alembic
```

Then, you can run the following command to initialize `alembic`.

```
alembic init migrations
```

This will create a `migrations` directory in the root directory of the project.

The `migrations` directory contains the following structure:

```
├── migrations
│   ├── README
│   
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
├── alembic.ini
```

Notice that it will also create a alembic.ini file in the root directory of the project.

```
[alembic]
# other configs
sqlalchemy.url = driver://user:pass@localhost/dbname
```

You need to specify the `sqlalchemy.url` setting in the `alembic.ini` file. This setting is used to connect to the database.

Then edit the `env.py` file and add the following line to the section [target_metadata]:

```
from src.database import SQLModel
target_metadata = SQLModel.metadata
```

Then open the `script.py.mako` file and add the following at the top

```
import sqlmodel
```

Now, we are ready to run our first migration.

```
alembic revision --autogenerate -m "Initial migration"
```

This command will create a new migration file in the `migrations/versions` directory.

Now, you can run the migration using the following command:

```
alembic upgrade head
```

This command will create the database and tables in the production environment.

But wait, we are not in the production environment yet. We are still in the development environment. So, we need to create a separate configuration file for the development environment.

## Create a separate configuration file for the development environment.

Create a `development.env` file in the root directory of the project with the following content:

```
ENV=development
DATABASE_URL=sqlite:///./test.db
# development.env
```

The `development.env` file contains the configuration settings for the development environment.

Now you can run the application using the following command:

```
uvicorn src.main:app --reload --env-file .development.env
```

This command will run the application in the development environment.

For local development, we need a local database. And we can use docker to create a local database. It will greatly improve the local development experience.

## Create a local database using Docker

Let’s see how we can use docker-compose to create a local database and run the local server from the same file.

This will allow us to run the application and the database using a single command.

First, install Docker and Docker Compose on your machine.

Then create the base `Dockerfile` in the root directory of the project with the following content:

```
FROM python:3.12.1-alpine3.18

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt
```

You can deploy your application anywhere with this docker file.

Then create a `docker-compose.yml` file in the root directory of the project with the following content:

```
version: '3.8'

services:
  db:
    image: postgres:13
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: test
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  app:
    build: .
    command: uvicorn src.main:app --reload --env-file .development.env --host 0.0.0.0 --port 8000
    volumes:
      - .:/app
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/test
    depends_on:
      - db

volumes:
    postgres_data:
```

The `db` service is used to create the database. It uses the `postgres:13` image. The `POSTGRES_USER`, `POSTGRES_PASSWORD`, and `POSTGRES_DB` settings are used to create the database.

The `app` service is used to run the application. It uses the `uvicorn` command to run the application. The `--env-file .development.env` setting is used to load the configuration settings from the `development.env` file.

## Database credentials

In this configuration, our local database URL will be `postgresql://user:password@localhost:5432/test`

But in the docker file, we used `DATABASE_URL=postgresql://user:password@db:5432/test` because in this context, the database is a service, and the host is `db`.

You can update the `.development.env` file to use the new database URL in case we want to access it from the local machine.

```
# .development.env
ENV=development
DATABASE_URL=postgresql://user:password@localhost:5432/test
```

Also, don’t forget to upgrade the `alembic.ini` file to use the new database url.

```
[alembic]
# other configs
sqlalchemy.url = postgresql://user:password@localhost:5432/test
```
Now, you can run the application and the database using the following command:

This command will create the database and run the application.

```
docker-compose up
```

Now, you can run your first migration on the local database using the following command:

```
alemibc revision --autogenerate -m "Initial migration"
```

This command will create a new migration file in the `migrations/versions` directory.

Now, you can run the migration using the following command:

```
alembic upgrade head
```

Now, if you visit your local database, you will see the `user` table.

## Create the user API

Now, let’s create the user API. We will create a router for the user domain.

Create a `router.py` file in the `src/api/user` directory with the following content:

```
from fastapi import APIRouter

router = APIRouter()

@router.get("/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]
```

The `router` object is an instance of the `APIRouter` class. It is used to define the routes for the user domain.

Now, you can add the user router to the main application using the following code:

```
from fastapi import FastAPI

from src.api.user.router import router as user_router

app = FastAPI()

app.include_router(user_router, prefix="/users", tags=["users"])
```

But we don’t want to deal with dummy data. Instead, we want to use the database to store and retrieve the users.

## Create the user service.

Create a `service.py` file in the `src/api/user` directory with the following content:

```
from fastapi import Depends
from src.database import get_session
from src.api.user.models import User
from sqlmodel import  select, Session


class UserService:
    def __init__(self, session: Session = Depends(get_session)) -> None:
        self.session = session

    def get_users(self):
        statement = select(User)
        users = self.session.exec(statement).all()
        return users
```
The `get_users` function is used to get the users from the database. Also, we are initializing the session in the constructor.

Now, you can use the `get_users` Function in the user router using the following code:

```
from fastapi import APIRouter, Depends

from src.api.user.service import UserService

router = APIRouter()

@router.get("/")
async def read_users(user_service : UserService = Depends()):
    users = user_service.get_users()
    return users
```

Notice that we are using the `Depends` function to inject the `UserService` object into the `read_users` function.

Now you can run the application and navigate to **http://localhost:8000/users** to see the users.

## Create a user.

Now, let’s create another function to create a user. But before that, we need to create the request and response models.

Create a new file named `schems.py` in the `src/api/user` directory with the following content:

```
from pydantic import BaseModel

class UserCreateInput(BaseModel):
    name: str
    email: str
    password: str
```

Then update the `service.py` file to include the `create_user` function.

```
def create_user(self, user_create_input):
        user = User(**user_create_input.model_dump())
        self.session.add(user)
        self.session.commit()
        self.session.refresh(user)
        return user
```

Finally, add the route.

```
from fastapi import APIRouter, Depends

from src.api.user.service import UserService

router = APIRouter()

@router.get("/")

async def read_users(user_service : UserService = Depends()):
    users = user_service.get_users()
    return users

@router.post("/")
async def create_user(user_create_input: UserCreateInput, user_service :UserService = Depends()):
    user = user_service.create_user(user_create_input)
    return user
```

Now, if you go to the terminal and send the following post request

```
curl -X 'POST' \
  'http://localhost:8000/users/' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "name": "Mohammad Faisal",
  "email": "mohammadfaisal1011@gmail.com",
  "password": "faisal"
}'
```

You will receive the following response with the success message!

## Get the details of a user.

Now, let’s create another function to get a user's details.

For this, we don’t need any request schema, as we will be using the user id to get the details.

```
@router.get("/{user_id}")
async def read_user(user_id: int, user_service : UserService = Depends()):
    user = user_service.get_user(user_id)
    return user
```

Then, remember to add the function to the service.

```
def get_by_id(self, user_id: int):
    statement = select(User).where(User.id == user_id)
    user = self.session.exec(statement).one()
    return user
```

But we need to handle the case when the user is not found.

```js
from sqlmodel import select, Session, SQLModel

class UserService:
    def get_by_id(self, user_id: int):
        statement = select(User).where(User.id == user_id)
        user = self.session.exec(statement).one_or_none()
        if user is None:
            raise Exception("User not found")
        return user
```

Now, this check will raise an exception if the user is not found.

But we need to handle this exception in the router.

```js
from fastapi import APIRouter, Depends, HTTPException

from src.api.user.service import UserService

router = APIRouter()

@router.get("/")

async def read_users(user_service : UserService = Depends()):
    users = user_service.get_users()
    return users

@router.post("/")
async def create_user(user_create_input: UserCreateInput, user_service :UserService = Depends()):
    user = user_service.create_user(user_create_input)
    return user

@router.get("/{user_id}")
async def read_user(user_id: int, user_service : UserService = Depends()):
    try:
        user = user_service.get_by_id(user_id)
        return user
    except Exception as e:
        raise HTTPException(status_code=404, detail="User not found")
```

Now, you will receive a 404 error if the user is not found.

### Update The User

Now, let’s create another function to update the user.

For this, we need to create a request schema.

```js
class UserUpdateInput(BaseModel):
    name: str
    email: str
    password: str
```

Then update the `service.py` file to include the `update_user` function.

```js
def update_user(self, user_id, user_update_input):
    statement = select(User).where(User.id == user_id)
    user = self.session.exec(statement).one()
    for key, value in user_update_input.dict().items():
        setattr(user, key, value)
    self.session.add(user)
    self.session.commit()
    self.session.refresh(user)
    return user
```

Finally, add the route.

```js
@router.put("/{user_id}")
async def update_user(user_id: int, user_update_input: UserUpdateInput,user_service : UserService = Depends()):
    try:
        user = user_service.get_by_id(user_id)
        return user
    except Exception as e:
        raise HTTPException(status_code=404, detail="User not found")
```

Now you can see that we have added the check for the user. If the user is not found, we will raise a 404 error.

Now, this is a duplicate code. We can move this check to the service.

We can use the concept of dependencies to create a dependency that will check if the user exists.

Create a new file named `dependencies.py` in the `src/api/user` directory with the following content:

```
from fastapi import HTTPException, Depends

from src.api.user.service import UserService

def get_user(user_id: int, user_service: UserService = Depends()):
    user = user_service.get_by_id(user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user
```

Then, update the `router.py` file to include the `get_user` dependency.

```
@router.get("/{user_id}")
async def read_user(
    user_id: int,
    user_service: UserService = Depends(),
    user: Mapping = Depends(get_user),
):
    user = user_service.get_by_id(user_id)
    return user
    
    
@router.put("/{user_id}")
async def update_user(
    user_id: int,
    user_update_input: UserUpdateInput,
    user_service: UserService = Depends(),
    user: Mapping = Depends(),
):
    user = user_service.update(user_id, user_update_input)
    return user
```

Now you can see that we are using the `get_user` dependency to check if the user exists. And you can do the same. Now, you don't have any duplications.

## Delete the user.

Now, let’s create another function to delete the user.

Then, update the `service.py` file to include the `delete_user` function.

```
def delete_user(self, user_id):
    statement = select(User).where(User.id == user_id)
    user = self.session.exec(statement).one()
    self.session.delete(user)
    self.session.commit()
    return user
```

Finlly, add the route.

```
@router.delete("/{user_id}")
async def delete_user(user_id: int, user_service : UserService =Depends()):
    user = user_service.delete(user_id)
    return user
```

That’s about it for today! Now you have a boilerplate project that you can just clone and start adding endpoints.

Hope you learned something new today!


**Have something to say? Get in touch with me via [LinkedIn](https://www.linkedin.com/in/56faisal/) or [Personal Website](https://www.mdfaisal.com/)**