postgres
databases
Starting with the fastapi-hello boilerplate, you can find an example API working with a database. All your application code should be written inside the ./src/ folder.
User model/entity.Using environment variables inside the .env file helps manage configuration settings in different environments (development, testing, production) without changing the code. FastAPI can use environment variables for configuration through libraries like python-dotenv.
.env.example file to create a .env file at the root of your project:1DATABASE_URL=postgresql+psycopg2://gitpod:postgres@localhost:5432/example 2 3# Add any other variables below 4SOME_VAR=SOME_VALUE
Routing in FastAPI is done through the use of APIRouter objects which can be included in the main application.
./src/endpoints/user.py:1from fastapi import APIRouter, Depends 2from sqlalchemy.orm import Session 3from typing import List 4from .models import User as UserModel 5from .database import get_db 6from .utils import APIException 7from pydantic import BaseModel, EmailStr 8 9router = APIRouter() 10 11# Serializers are used to validate the incoming request body 12# Here you determine which fields are required and their types 13class CreateSerializer(BaseModel): 14 password: str = Field(..., min_length=3, max_length=50) 15 email: EmailStr 16 is_active: bool 17 18# Serializers are also used to format the outgoing response body 19class UserSmallSerializer(BaseModel): 20 email: str 21 is_active: bool 22 23 class Config: 24 from_attributes = True 25 26@router.get("/users/") 27def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)): 28 users = db.query(UserModel).offset(skip).limit(limit).all() 29 return [UserSmallSerializer.model_validate(user) for user in users] 30 31@router.get("/users/{user_id}") 32def read_user(user_id: int, db: Session = Depends(get_db)): 33 user = db.query(UserModel).filter(UserModel.id == user_id).first() 34 if user is None: 35 raise APIException(status_code=404, detail="User not found") 36 return UserSmallSerializer.model_validate(user) 37 38@router.post("/users/") 39def create_user(user: CreateSerializer, db: Session = Depends(get_db)): 40 db_user = UserModel(username=user.username, email=user.email, age=user.age) 41 db.add(db_user) 42 db.commit() 43 return UserSmallSerializer.model_validate(db_user) 44 45@router.put("/users/{user_id}", response_model=UserSmallSerializer) 46def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)): 47 db_user = db.query(UserModel).filter(UserModel.id == user_id).first() 48 if db_user is None: 49 raise APIException(status_code=404, detail="User not found") 50 for key, value in user.dict().items(): 51 setattr(db_user, key, value) 52 db.commit() 53 return UserSmallSerializer.model_validate(db_user) 54 55@router.delete("/users/{user_id}", response_model=UserSmallSerializer) 56def delete_user(user_id: int, db: Session = Depends(get_db)): 57 db_user = db.query(UserModel).filter(UserModel.id == user_id).first() 58 if db_user is None: 59 raise APIException(status_code=404, detail="User not found") 60 db.delete(db_user) 61 db.commit() 62 return UserSmallSerializer.model_validate(db_user)
🔥 All the python files inside
./src/endpointswill be automatically included as routes in your API, there is no need to use theapp.include_routerfunction.
In API development, a concept called "Serialization" is used to receive and return data:
In FastAPI, you can create a Serializer using a library called "Pydantic" and these serializers are called Pydantic Models.
The following serializer called CreateSerializer was defined to validate the incoming data payload used during the POST /user endpoint that creates a new user, the endpoint payload must contain a password, email and is_active boolean.
1# Serializers are used to validate the incoming request body 2# Here you determine which fields are required and their types 3class CreateSerializer(BaseModel): 4 password: str = Field(..., min_length=3, max_length=50) 5 email: EmailStr 6 is_active: bool
We have to specify our CreateSerializer class as the first parameter of the function that handles the POST method, in this case the create_user function and FastAPI will do the validations:
1@router.post("/users/") 2# ⬇️ here we add the CreateSerializer 3def create_user(incoming_payload: CreateSerializer, db: Session = Depends(get_db)): 4 5 # The incoming_payload has already been validated, and you can "trust" it. 6 db_user = UserModel( 7 password=incoming_payload.password, 8 email=incoming_payload.email, 9 is_active=incoming_payload.is_active 10 ) 11 db.add(db_user) 12 db.commit() 13 return UserSmallSerializer.model_validate(db_user)
Serialization is handled by Pydantic models which automatically convert Python objects to JSON text.
./src/endpoints/user.py:
1@router.post("/users/", response_model=UserSmallSerializer) 2def create_user(user: UserCreate, db: Session = Depends(get_db)): 3 db_user = UserModel(username=user.username, email=user.email, age=user.age) 4 db.add(db_user) 5 db.commit() 6 return UserSmallSerializer.model_validate(db_user)
For database operations we use SQLAlchemy.
Define a database model in ./src/models.py:
1from sqlalchemy import Column, Integer, String 2from .database import Base 3 4class User(Base): 5 __tablename__ = "users" 6 id = Column(Integer, primary_key=True, index=True) 7 username = Column(String, unique=True, index=True) 8 email = Column(String, unique=True, index=True) 9 age = Column(Integer)
Here are some different examples on creating, deleting and updating a user:
1# Create a new SQLAlchemy user instance 2db_user = UserModel(username=user.username, email=user.email, age=user.age) 3db.add(db_user) # Add the user to the session 4db.commit() # Commit the session to save the user in the database 5 6# Delete a user by id 7db_user = db.query(UserModel).filter(UserModel.id == user_id).first() 8if db_user is None: raise APIException(status_code=404, detail="User not found") 9db.delete(db_user) # Delete the user from the session 10db.commit() # Commit the session to remove the user from the database 11 12# Update the user by id 13db_user = db.query(UserModel).filter(UserModel.id == user_id).first() 14if db_user is None: raise APIException(status_code=404, detail="User not found") 15# update the fields you need to update, for example: 16db_user.username = "some_new_username" 17# Commit the session to save the changes in the database 18db.commit() 19 20# Get all users with more than 18 yrs old 21users = db.query(UserModel).filter(UserModel.age > 18).all()
This guide covers the basic setup and use of FastAPI with environment variables, routing (including CRUD operations), validations, serialization using Pydantic models, and database operations with SQLAlchemy. With these components in place, you can build scalable and robust web applications using FastAPI.