FASTAPI
Database Connection Steps
Installing the required packages.
Configuring the database connection.
Creating database models.
Creating database sessions.
Using the database inside FastAPI endpoints.
[Link] Required Packages (Install FastAPI and database
dependencies)
pip install fastapi uvicorn sqlalchemy psycopg2-binary or pymysql
What these packages do
Package Purpose
fastapi Web framework
uvicorn ASGI server
sqlalchemy ORM (Database Toolkit)
pymysql MysqlSQL driver
2. Create the following structure:
project/
│
├── [Link]
├── [Link]
├── [Link]
├── [Link]
└── [Link]
3. [Link]
from sqlalchemy import create_engine
from [Link] import sessionmaker, declarative_base
MYSQL_USER = "root"
MYSQL_PASSWORD = "root"
MYSQL_HOST = "localhost"
MYSQL_PORT = "3306"
MYSQL_DB = "fastdb" # database name created in Mysql
DATABASE_URL = (
f"mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}"
f"@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DB}"
)
engine = create_engine(DATABASE_URL, echo=True) # connection object
SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=engine
) # cursor object
Base = declarative_base()
Explanation
create_engine → Creates a connection interface to the database.
sessionmaker → Creates a factory for database sessions.
declarative_base → Creates a base class that your ORM models will inherit from.
The engine is SQLAlchemy's core connection manager.
engine = create_engine(DATABASE_URL)
acts like connection object
A session is used to interact with the database. (acts like cursor object)
SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=engine
)
bind=engine :- Associates all sessions created by SessionLocal with the engine.
autocommit=False :- Changes are not automatically saved.
autoflush=False :- Prevents SQLAlchemy from automatically sending pending changes to the
database before queries.
Base = declarative_base()
This creates a parent class for all ORM models.
4. [Link]
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100))
email = Column(String(100), unique=True, index=True)
Explanation:-
from sqlalchemy import Column, Integer, String
These are SQLAlchemy column types and utilities:
Column → Defines a table column.
Integer → Integer data type.
String → Text/VARCHAR data type.
Base = declarative_base()All ORM models should inherit from Base.
By inheriting from Base, SQLAlchemy knows this class should be mapped to a database table.
5. [Link]
from pydantic import BaseModel
# POST
class UserCreate(BaseModel):
name: str
email: str
class UserUpdate(BaseModel):
name : str
email :str
# GET
class UserResponse(UserCreate):
id: int
class Config:
orm_mode = True
Explanation :-
from pydantic import BaseModel
BaseModel is the parent class for all Pydantic models.
Pydantic provides:
Data validation
Type checking
Automatic conversion
API documentation generation in FastAPI
Example:
class UserCreate(BaseModel):
name: str
email: str
If someone sends:
{
"name": "Alice",
"email": "alice@[Link]"
}
Pydantic validates it automatically.
6. [Link] (optional)
from [Link] import Session
import models, schema
# code of POST endpoint
def create_user(db: Session, user: [Link]):
db_user = [Link](
name=[Link],
email=[Link]
)
[Link](db_user)
[Link]()
[Link](db_user)
return db_user
# code for GET endpoint
def get_users(db: Session):
return [Link]([Link]).all()
7. [Link]
from fastapi import FastAPI, Depends, HTTPException
from [Link] import Session
import models
from database import engine, SessionLocal
import schema, qry
app = FastAPI()
# Create tables
[Link].create_all(bind=engine)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
[Link]()
# GET - All users
@[Link]("/users/", response_model=list[[Link]])
def get_users(db: Session = Depends(get_db)):
return [Link]([Link]).all()
# GET - Single user
@[Link]("/users/{user_id}", response_model=[Link])
def get_user(user_id: int, db: Session = Depends(get_db)):
user = [Link]([Link]).filter([Link] == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
# POST - Create user
@[Link]("/users/", response_model=[Link])
def create_user(user: [Link], db: Session = Depends(get_db)):
db_user = [Link](name=[Link], email=[Link])
[Link](db_user)
[Link]()
[Link](db_user)
return db_user
# PUT - Update user
@[Link]("/users/{user_id}", response_model=[Link])
def update_user(user_id: int, user_data: [Link], db: Session =
Depends(get_db)):
user = [Link]([Link]).filter([Link] == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
# update data from new details to existing record
[Link] = user_data.name
[Link] = user_data.email
[Link]()
[Link](user)
return user
# DELETE - Delete user
@[Link]("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
user = [Link]([Link]).filter([Link] == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
[Link](user)
[Link]()
return {"message": "User deleted successfully"}
Explanation
[Link].create_all(bind=engine)
Check models
↓
Generate SQL
↓
Create tables if missing
def get_db(): -> This function creates and manages database sessions.
response_model=list[[Link]]
"Return a list of UserResponse objects."