Introduction
In the fast-evolving world of web development, choosing the right tools is crucial for building robust and efficient applications. This article will guide you through the process of setting up FastAPI, a modern, fast, web framework for building APIs with Python, along with SQLAlchemy, a powerful SQL toolkit, and MySQL, a popular open-source relational database management system. By the end of this tutorial, you'll have a solid foundation for developing high-performance web applications.
Guide Overview:
This guide outlines the professional setup of FastAPI in conjunction with SQLAlchemy, utilizing MySQL as the chosen database. By following these steps, you will establish a robust foundation for developing sophisticated and high-performance web applications.
Installation Steps:
To initiate the setup, execute the following commands to install the necessary dependencies:
1. pip install fastapi "uvicorn[standard]"
2. pip install python-dotenv
3. pip install sqlalchemy
4. pip install pymysql
This will install Fastapi and uvicorn to work as the server.
Requirements:
- MySql Database
Folder setup:
/app/
- __init__.py
- main.py
- database.py
- models.py
- schema.py
- crud.py
/.env
/requirements.txt
First Let’s create a database engine : app/database.py
app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from dotenv import load_dotenv
load_dotenv()
import os
DB_URL = DB_URL = os.getenv("DB_URL")
engine = create_engine(DB_URL,echo=True)
SessionLocal = sessionmaker(autocommit=False,autoflush=False, bind=engine)
Base = declarative_base()
This is the part where we are connecting to MYSQL database
Create database model:
app/models.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer,primary_key=True,index=True)
name = Column(String(255),index=True)
email = Column(String(255), unique=True, index=True)
todos = relationship("Todo",back_populates="owner")
is_active = Column(Boolean,default=False)
class Todo(Base):
__tablename__ = "todos"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255), index=True)
description = Column(String(255), index=True)
owner_id = Column(Integer, ForeignKey("users.id"))
owner = relationship("User",back_populates="todos")
Create Pydantic models/schemas:
app/schemas.py
from pydantic import BaseModel
class TodoBase(BaseModel):
title : str
description : str | None = None
class TodoCreate(TodoBase):
pass
class Todo(TodoBase):
id : int
owner_id : int
class Config:
orm_mode = True
class UserBase(BaseModel):
email: str
name: str
class UserCreate(UserBase):
pass
class User(UserBase):
id : int
is_active : bool
todos : list[Todo] = []
class Config:
orm_model = True
Crud Utils:
app/crud.py
from sqlalchemy.orm import Session
from . import models,schemas
def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip:int=0, limit:int=100):
# return db.query(models.User).offset(skip).limit(limit).all()
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user:schemas.UserCreate):
db_user = models.User(email=user.email,
name=user.name)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def get_todos(db: Session, skip:int=0, limit: int=100):
return db.query(models.Todo).offset(skip).limit(limit).all()
def create_user_todo(db:Session, todo:schemas.TodoCreate, user_id : int):
db_todo = models.Todo(**todo.model_dump(),owner_id=user_id )
db.add(db_todo)
db.commit()
db.refresh(db_todo)
return db_todo
# NOTE :
# - add that instance object to your database session.
# - commit the changes to the database (so that they are saved).
# - refresh your instance (so that it contains any new data from the database, like the generated ID).
FastAPI Entry point
app/main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import crud,models, schemas
from .database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
#Dependency
def get_db():
db = SessionLocal()
try :
yield db
finally:
db.close()
@app.post("/users/",response_model=schemas.User)
def post_user(user:schemas.UserCreate, db:Session=Depends(get_db)):
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return crud.create_user(db=db,user=user)
@app.get("/users/", response_model=list[schemas.User])
def get_users(skip:int=0, limit:int=0, db:Session=Depends(get_db)):
users = crud.get_users(db,skip=skip,limit=limit)
return users
@app.get("/users/{user_id}/",response_model=schemas.User)
def get_user(user_id:int, db:Session=Depends(get_db)):
db_user = crud.get_user(db,user_id =user_id )
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
@app.post("/users/{user_id}/todos/",response_model=schemas.Todo)
def post_todo_for_user(user_id:int, todo:schemas.TodoCreate, db:Session=Depends(get_db)):
return crud.create_user_todo(db=db,user_id=user_id, todo=todo)
@app.get("/todos/", response_model=list[schemas.Todo])
def get_todos(skip:int=0,limit:int=100,db:Session=Depends(get_db)):
todos = crud.get_todos(db,skip=skip,limit=limit)
return todos
Environment Variables:
.env
DB_URL = "mysql+pymysql://{db_username}:{db_password}@localhost:3306/{db_name}"
Result:
uvicorn app.main:app --reload
Conclusion
Congratulations! You've successfully set up a FastAPI application integrated with SQLAlchemy and MySQL.
Thank you 😊
Top comments (0)