FastAPI & SQLAlchemy: MySQL Setup Guide
Tags: python, fastapi, mysql, sqlalchemy
Published: 2023-12-15 09:51:32

## 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:

```python
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:

1. MySql Database

## Folder setup:

```python
/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`

```python
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`

```python
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`

```python
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`

```python
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`

```python
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`

```python
DB_URL = "mysql+pymysql://{db_username}:{db_password}@localhost:3306/{db_name}"
```

## Result:

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


![FastAPI](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/yzh2fdzlxi4d64835m2o.png)


## Conclusion

Congratulations! You've successfully set up a FastAPI application integrated with SQLAlchemy and MySQL.

Thank you 😊