To read more articles like this, visit my 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…
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.
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:
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:
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:
[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:
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.
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.
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.
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.
class UserUpdateInput(BaseModel):
name: str
email: str
password: str
Then update the service.py
file to include the update_user
function.
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.
@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 or Personal Website
Top comments (0)