🚀 Lesson 6 — Databases in FastAPI (SQL + NoSQL, SQLAlchemy ORM, Async DB, CRUD)
This is one of the most IMPORTANT and most asked in interviews because almost all real apps talk to a database.
By the end of this lesson, you will know:
✔ SQL databases (PostgreSQL, MySQL, SQLite)
✔ NoSQL databases (MongoDB)
✔ SQLAlchemy ORM (industry standard)
✔ Async SQLAlchemy + PostgreSQL (modern stack)
✔ Complete CRUD operations
✔ Relationships (one-to-many)
✔ Dependency injection for DB sessions
✔ Industry-level project structure
Let’s begin. 🔥
🎯 What You Will Learn Today
Part A — SQL Databases
✔ SQLAlchemy ORM
✔ Database models
✔ DB session
✔ CRUD operations
✔ Relationships
Part B — Async Databases
✔ Async SQLAlchemy
✔ PostgreSQL connection
✔ Async CRUD
Part C — NoSQL (MongoDB)
✔ FastAPI + Motor (Async Mongo)
🧠 Part A — SQL DATABASE + SQLAlchemy (Synchronous)
This is the most common setup for:
- E-commerce
- Payments
- Banking
- Job portals
- SaaS backend
📌 1. Install SQLAlchemy + SQLite
pip install sqlalchemy
🧱 2. Create Database Model (User Table)
models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String)
email = Column(String, unique=True, index=True)
🧵 3. Create Engine & Session
database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
📦 4. Dependency for DB Session
main.py
from fastapi import Depends
from database import SessionLocal
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
This is industry pattern.
🧩 5. Create Pydantic Schemas
schemas.py
from pydantic import BaseModel
class UserCreate(BaseModel):
name: str
email: str
class UserRead(BaseModel):
id: int
name: str
email: str
class Config:
orm_mode = True
orm_mode=True is required so FastAPI can convert SQL rows → JSON.
🛠 6. CRUD APIs
👉 Create User
@app.post("/users", response_model=UserRead)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
new_user = User(name=user.name, email=user.email)
db.add(new_user)
db.commit()
db.refresh(new_user)
return new_user
👉 Read User
@app.get("/users/{user_id}", response_model=UserRead)
def get_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(User).filter(User.id == user_id).first()
return user
👉 Read All Users
@app.get("/users", response_model=list[UserRead])
def list_users(db: Session = Depends(get_db)):
return db.query(User).all()
👉 Update User
@app.put("/users/{user_id}", response_model=UserRead)
def update_user(user_id: int, data: UserCreate, db: Session = Depends(get_db)):
user = db.query(User).filter(User.id == user_id).first()
user.name = data.name
user.email = data.email
db.commit()
db.refresh(user)
return user
👉 Delete User
@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(User).get(user_id)
db.delete(user)
db.commit()
return {"msg": "deleted"}
🔗 7. One-to-Many Relationship Example
Example:
1 User → many Orders
models.py
class Order(Base):
__tablename__ = "orders"
id = Column(Integer, primary_key=True)
product = Column(String)
user_id = Column(Integer, ForeignKey("users.id"))
This is used in:
- Amazon orders
- Swiggy food order items
- Sales platforms
🚀 Part B — ASYNC SQL DATABASE (PostgreSQL + Async SQLAlchemy)
This is modern, high-performance, and used in:
- High-traffic AI APIs
- Microservices
- Cloud-native apps
Install:
pip install sqlalchemy[asyncio] asyncpg
⚡ 1. Async Engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
⚡ 2. Dependency
async def get_db():
async with AsyncSessionLocal() as session:
yield session
⚡ 3. Async CRUD Example
Create User
@app.post("/users")
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
new = User(name=user.name, email=user.email)
db.add(new)
await db.commit()
await db.refresh(new)
return new
🌐 Part C — MongoDB (Async NoSQL)
Used in:
- Logging
- Chat apps
- AI vector DBs
- IoT platforms
- Real-time dashboards
Install:
pip install motor
1. Connect to MongoDB
from motor.motor_asyncio import AsyncIOMotorClient
client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client.mydb
2. Insert Document
@app.post("/product")
async def add_product(product: dict):
result = await db.products.insert_one(product)
return {"id": str(result.inserted_id)}
3. Get Document
@app.get("/product/{id}")
async def get_product(id: str):
product = await db.products.find_one({"_id": ObjectId(id)})
product["_id"] = str(product["_id"])
return product
📌 Lesson 6 Summary
In this lesson, you learned:
✔ SQL
- SQLAlchemy ORM
- DB models
- Sessions & dependencies
- CRUD operations
- One-to-many relationships
✔ Async SQL
- PostgreSQL async
- Async CRUD
- Enterprise pattern
✔ NoSQL
- MongoDB + Motor
- Async CRUD
This is exactly what companies expect in a backend/ML engineer interview.
🚀 Ready for Lesson 7 — FastAPI for AI/ML Model Serving (HuggingFace, PyTorch, TensorFlow, Transformers)?
Shall I continue with Lesson 7 now?