Building a Full-Stack Todo Web App with FastAPI and PostgreSQL
8 min readFeb 6, 2024
In this blog post, we’ll guide you through the process of building a full-stack Todo web app using FastAPI as the backend framework and PostgreSQL as the database. FastAPI’s speed and simplicity, combined with PostgreSQL’s reliability, make for a powerful and scalable combination.
1. Setting Up Your Development Environment
1.1 Install Python.
If you don't already have Python installed, you can download it from the Python website.
Create a new folder called ‘todo_app’ and in this folder, make Venv
python -m venv venv
source venv/bin/activate1.2 Installing FastAPI and setting up a new project.
pip install fastapi==0.100.11.3 Configuring PostgreSQL as the database for your project.
You can use any database you want, like SQLite, PostgreSQL, and MySql.
2. Building the Backend with FastAPI
This is a directory structure of the project.
main.py
from fastapi import FastAPI
import uvicorn
import models
from database import engine
from routers import auth, todos, users, address
from starlette.staticfiles import StaticFiles
app = FastAPI()
models.Base.metadata.create_all(bind=engine)
app.mount('/static', StaticFiles(directory='static'), name='static')
app.include_router(todos.router)
app.include_router(auth.router)
app.include_router(address.router)
app.include_router(users.router)
if __name__ == '__main__':
uvicorn.run("main:app", reload=True)In this file, include all the routers that were created in the router folder
database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# SQLALCHEMY_DATABASE_URL = "sqlite:///./todos.db"
SQLALCHEMY_DATABASE_URL = "postgresql://postgres:12345678@localhost/Todos"
# SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:12345678@localhost:3306/todos"
# , connect_args={"check_same_thread": False}
engine = create_engine(
SQLALCHEMY_DATABASE_URL
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()In the above code, we created an engine using create_engine to connect the database.
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey, Double
from sqlalchemy.orm import relationship
from database import Base
class Users(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, index=True)
email = Column(String(200), unique=True, index=True)
username = Column(String(200), unique=True, index=True)
first_name = Column(String(200))
last_name = Column(String(225))
hashed_password = Column(String(225))
is_active = Column(Boolean, default=True)
phone_number = Column(String(225))
address_id = Column(Integer, ForeignKey('address.id'), nullable=True)
todos = relationship('Todos', back_populates='owner')
address = relationship('Address', back_populates='user_address')
class Todos(Base):
__tablename__ = 'todos'
id = Column(Integer, primary_key=True, index=True)
title = Column(String(225))
amount = Column(Double)
description = Column(String(225))
owner_id = Column(Integer, ForeignKey('users.id'))
owner = relationship('Users', back_populates='todos')
class Address(Base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True, index=True)
apt_num = Column(Integer) # alembic
address1 = Column(String(255))
address2 = Column(String(255))
city = Column(String(255))
state = Column(String(255))
country = Column(String(255))
postalcode = Column(String(255))
user_address = relationship('Users', back_populates='address')Users for storing different user information, Todos for storing their todos, and Address for storing user addresses.
routers/user.py
import sys
from pydantic import BaseModel
from sqlalchemy.orm import Session
import models
from database import engine, SessionLocal
from fastapi import Depends, APIRouter, status
from routers.auth import get_current_user, get_user_exception, verify_password, get_password_hash
sys.path.append('..')
router = APIRouter(
prefix='/users',
tags=['Users'],
responses={
status.HTTP_404_NOT_FOUND: {
'description': 'Not found'
}
}
)
models.Base.metadata.create_all(bind=engine)
def get_db():
try:
db = SessionLocal()
yield db
finally:
db.close()
class UserVerification(BaseModel):
username: str
password: str
new_password: str
class PhoneNumber(BaseModel):
phone_number: str
@router.get('/')
async def read_all(db: Session = Depends(get_db)):
return db.query(models.Users).all()
@router.get('/user/{user_id}')
async def user_by_path(user_id: int, db: Session = Depends(get_db)):
user_model = db.query(models.Users).filter(models.Users.id == user_id).first()
if user_model is not None:
return user_model
return 'Invalid User'
@router.get('/user/')
async def user_by_query(user_id: int, db: Session = Depends(get_db)):
user_model = db.query(models.Users).filter(models.Users.id == user_id).first()
if user_model is not None:
return user_model
return 'Invalid User'
@router.put('/user/password')
async def user_password_change(user_verification: UserVerification,
user: dict = Depends(get_current_user),
db: Session = Depends(get_db)):
if user is None:
raise get_user_exception()
user_model = db.query(models.Users).filter(models.Users.id == user.get('id')).first()
if user_model is not None:
if (user_verification.username == user_model.username and
verify_password(user_verification.password, user_model.hashed_password)):
user_model.hashed_password = get_password_hash(user_verification.new_password)
db.add(user_model)
db.commit()
return 'successful'
return 'Invalid user or request'
@router.delete('/user')
async def delete_user(db: Session = Depends(get_db),
user: dict = Depends(get_current_user)):
if user is None:
raise get_user_exception()
user_model = db.query(models.Users).filter(models.Users.id == user.get('id')).first()
if user_model is None:
return 'Invalid user or request'
db.query(models.Users).filter(models.Users.id == user.get('id')).delete()
db.commit()
return 'Delete successful'
@router.put('/user/phone')
async def add_phone_number(phone_number: PhoneNumber,
db: Session = Depends(get_db),
user: dict = Depends(get_current_user)):
if user is None:
raise get_user_exception()
user_model = db.query(models.Users).filter(models.Users.id == user.get('id')).first()
if user_model is not None:
user_model.phone_number = phone_number.phone_number
db.add(user_model)
db.commit()
return 'Phone number updated successfully'
return 'Invalid user or request'routers/todos.py
import sys
from typing import Optional
from fastapi import Depends, HTTPException, APIRouter, status, Request, Form
from pydantic import BaseModel, Field
import models
from database import engine, SessionLocal
from sqlalchemy.orm import Session
from routers.auth import get_current_user, get_user_exception
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from starlette.responses import RedirectResponse
sys.path.append('..')
router = APIRouter(
tags=['Todos'],
responses={
status.HTTP_404_NOT_FOUND: {
'description': 'Not found'
}
}
)
models.Base.metadata.create_all(bind=engine)
templates = Jinja2Templates(directory='templates')
def get_db():
try:
db = SessionLocal()
yield db
finally:
db.close()
class Todo(BaseModel):
title: str
description: Optional[str]
priority: int = Field(gt=0, lt=6, description='The priority must be between 1-5')
complete: bool
@router.get('/', response_class=HTMLResponse)
async def read_all_by_user(request: Request,
db: Session = Depends(get_db)):
user = await get_current_user(request)
if user is None:
return RedirectResponse(url='/auth', status_code=status.HTTP_302_FOUND)
todos = (db.query(models.Todos)
.filter(models.Todos.owner_id == user.get('id'))
.order_by(models.Todos.id).all())
return templates.TemplateResponse('todo/home.html', {'request': request, 'todos': todos, 'user': user})
@router.get('/add-todo', response_class=HTMLResponse)
async def add_new_todo(request: Request):
user = await get_current_user(request)
if user is None:
return RedirectResponse(url='/auth', status_code=status.HTTP_302_FOUND)
return templates.TemplateResponse('todo/add_todo.html', {'request': request, 'user': user})
@router.post('/add-todo', response_class=HTMLResponse)
async def create_todo_commit(request: Request,
title: str = Form(...),
description: str = Form(...),
priority: int = Form(...),
db: Session = Depends(get_db)):
user = await get_current_user(request)
if user is None:
return RedirectResponse(url='/auth', status_code=status.HTTP_302_FOUND)
todo_model = models.Todos()
todo_model.title = title
todo_model.description = description
todo_model.priority = priority
todo_model.complete = False
todo_model.owner_id = user.get('id')
db.add(todo_model)
db.commit()
return RedirectResponse(url='/', status_code=status.HTTP_302_FOUND)
@router.get('/edit-todo/{todo_id}', response_class=HTMLResponse)
async def edit_todo(todo_id: int,
request: Request,
db: Session = Depends(get_db)):
user = await get_current_user(request)
if user is None:
return RedirectResponse(url='/auth', status_code=status.HTTP_302_FOUND)
todo = (db.query(models.Todos).filter(models.Todos.id == todo_id)
.filter(models.Todos.owner_id == user.get('id')).first())
return templates.TemplateResponse('todo/edit_todo.html', {'request': request, 'todo': todo, 'user': user})
@router.post('/edit-todo/{todo_id}', response_class=HTMLResponse)
async def edit_todo_commit(request: Request,
todo_id: int,
title: str = Form(...),
description: str = Form(...),
complete: bool = Form(False),
priority: int = Form(...),
db: Session = Depends(get_db)
):
user = await get_current_user(request)
if user is None:
return RedirectResponse(url='/auth', status_code=status.HTTP_302_FOUND)
todo_model = (db.query(models.Todos).filter(models.Todos.id == todo_id)
.filter(models.Todos.owner_id == user.get('id')).first())
todo_model.title = title
todo_model.description = description
todo_model.priority = priority
todo_model.complete = complete
db.add(todo_model)
db.commit()
return RedirectResponse(url='/', status_code=status.HTTP_302_FOUND)
@router.get('/delete/{todo_id}', response_class=HTMLResponse)
async def delete_todo(request: Request, todo_id: int, db: Session = Depends(get_db)):
user = await get_current_user(request)
if user is None:
return RedirectResponse(url='/auth', status_code=status.HTTP_302_FOUND)
todo_model = (db.query(models.Todos).filter(models.Todos.id == todo_id)
.filter(models.Todos.owner_id == user.get('id')).first())
if todo_model is None:
return RedirectResponse(url='/todos', status_code=status.HTTP_302_FOUND)
(db.query(models.Todos).filter(models.Todos.id == todo_id)
.filter(models.Todos.owner_id == user.get('id')).delete())
db.commit()
return RedirectResponse(url='/', status_code=status.HTTP_302_FOUND)
@router.get('/complete/{todo_id}', response_class=HTMLResponse)
async def complete_todo(request: Request, todo_id: int, db: Session = Depends(get_db)):
user = await get_current_user(request)
if user is None:
return RedirectResponse(url='/auth', status_code=status.HTTP_302_FOUND)
todo_model = (db.query(models.Todos).filter(models.Todos.id == todo_id)
.filter(models.Todos.owner_id == user.get('id')).first())
todo_model.complete = not todo_model.complete
db.add(todo_model)
db.commit()
return RedirectResponse(url='/', status_code=status.HTTP_302_FOUND)The above route is for creating, reading, updating, and deleting todos.
routers/auth.py
import sys
from fastapi import Depends, HTTPException, APIRouter, Request, Response
from passlib.context import CryptContext
from pydantic import BaseModel
from typing import Optional
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from starlette import status
import models
from database import engine, SessionLocal
from models import Users
from jose import JWTError, jwt
from datetime import datetime, timedelta
import time
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from starlette.responses import RedirectResponse
sys.path.append('..')
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
class CreateUser(BaseModel):
username: str
email: Optional[str]
first_name: str
last_name: str
password: str
phone_number: Optional[str]
bcrypt_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
models.Base.metadata.create_all(bind=engine)
templates = Jinja2Templates(directory='templates')
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
router = APIRouter(
prefix='/auth',
tags=['Auth'],
responses={
status.HTTP_401_UNAUTHORIZED: {
'user': 'Not authorized'
}
}
)
class LoginForm:
def __init__(self, request: Request):
self.request: Request = request
self.username: Optional[str] = None
self.password: Optional[str] = None
async def create_oauth_form(self):
form = await self.request.form()
self.username = form.get('email')
self.password = form.get('password')
class RegisterForm:
def __init__(self, request: Request):
self.request: Request = request
self.email: Optional[str] = None
self.username: Optional[str] = None
self.password: Optional[str] = None
self.password2: Optional[str] = None
self.firstname: Optional[str] = None
self.lastname: Optional[str] = None
async def create_register_form(self):
form = await self.request.form()
self.email = form.get('email')
self.username = form.get('username')
self.password = form.get('password')
self.password2 = form.get('password2')
self.firstname = form.get('firstname')
self.lastname = form.get('lastname')
def get_db():
try:
db = SessionLocal()
yield db
finally:
db.close()
def verify_password(plain_password, hashed_password):
return bcrypt_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return bcrypt_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(request: Request):
try:
token = request.cookies.get('access_token')
if token is None:
return None
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
user_id: int = payload.get("id")
if username is None or user_id is None:
print("us")
raise get_user_exception()
return {'username': username, 'id': user_id}
except JWTError:
raise get_user_exception()
def authenticate_user(db, username: str, password: str):
user = db.query(Users).filter(Users.username == username).first()
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
@router.post('/token')
async def login_for_access_token(response: Response,
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)):
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
return False # token_exception()
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username, 'id': user.id},
expires_delta=access_token_expires)
response.set_cookie(key='access_token', value=access_token, httponly=True)
return True # {"access_token": access_token, "token_type": "bearer"}
@router.get('/', response_class=HTMLResponse)
async def auth_page(request: Request):
user = await get_current_user(request)
if user is not None:
return RedirectResponse(url='/', status_code=status.HTTP_302_FOUND)
return templates.TemplateResponse('user/login.html', {'request': request})
@router.post('/', response_class=HTMLResponse)
async def login(request: Request, db: Session = Depends(get_db)):
try:
form = LoginForm(request)
await form.create_oauth_form()
response = RedirectResponse(url='/', status_code=status.HTTP_302_FOUND)
validate_user_cookie = await login_for_access_token(response=response, form_data=form, db=db)
if not validate_user_cookie:
msg = 'Incorrect Username or Password'
return templates.TemplateResponse('user/login.html', {'request': request, 'msg': msg})
return response
except HTTPException:
msg = 'Unknown Error'
return templates.TemplateResponse('user/login.html', {'request': request, 'msg': msg})
@router.get('/logout', response_class=HTMLResponse)
async def logout(request: Request):
msg = 'Logout Successful'
response = templates.TemplateResponse('user/login.html', {'request': request, 'msg': msg})
response.delete_cookie(key='access_token')
return response
@router.get('/register', response_class=HTMLResponse)
async def register_page(request: Request):
user = await get_current_user(request)
if user is not None:
return RedirectResponse(url='/', status_code=status.HTTP_302_FOUND)
return templates.TemplateResponse('user/register.html', {'request': request})
@router.post('/register')
async def create_new_user(request: Request, db: Session = Depends(get_db)):
form = RegisterForm(request)
await form.create_register_form()
user = Users()
user.email = form.email
user.username = form.username
user.first_name = form.firstname
user.last_name = form.lastname
user.hashed_password = get_password_hash(form.password)
# user.phone_number = form.phonenumber
user.is_active = True
db.add(user)
db.commit()
msg = 'User successfully created'
response = templates.TemplateResponse('user/login.html', {'request': request, 'msg': msg})
return response
def get_user_exception():
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
return credentials_exception
def token_exception():
token_exception_response = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
return token_exception_responseThis is user authentication, for getting the current user, creating a new user, logout, login, etc.
I use alembic for version management you can check out this also, I use HTML, CSS, and JavaScript for the front-end.
End-to-end code is provided in the GitHub repo.
By the end of this tutorial, you’ll have a fully functional Todo web app with a backend powered by FastAPI and a PostgreSQL database. This comprehensive guide aims to help both beginners and experienced developers streamline their full-stack development workflow using these powerful technologies. Get ready to embark on an exciting journey of building and deploying your feature-rich web application!
Feel free to connect:
LinkedIN : https://www.linkedin.com/in/gopalkatariya44/
Github : https://github.com/gopalkatariya44/
Instagram : https://www.instagram.com/_gk_44/
Twitter: https://twitter.com/GopalKatariya44
Youtube: https://youtube.com/@gopalkatariya44
Thanks 😊 !
