CRUD API with FastAPI + SQLAlchemy
Create, Read, Update, Delete users — the four operations every backend needs. SQLite + Pydantic + SQLAlchemy.
🚀 Mini Project · 2 filesProject files
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import sessionmaker, declarative_base, Session
import uvicorn
# --- DB setup ---
engine = create_engine('sqlite:///users.db', connect_args={'check_same_thread': False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String, unique=True)
email = Column(String)
Base.metadata.create_all(engine)
# --- Pydantic schema ---
class UserIn(BaseModel):
name: str
email: str
class UserOut(BaseModel):
id: int
name: str
email: str
class Config: from_attributes = True
# --- FastAPI ---
app = FastAPI()
def get_db():
db = SessionLocal()
try: yield db
finally: db.close()
@app.post('/users', response_model=UserOut)
def create(u: UserIn, db: Session = Depends(get_db)):
user = User(name=u.name, email=u.email)
db.add(user); db.commit(); db.refresh(user)
return user
@app.get('/users', response_model=list[UserOut])
def list_all(db: Session = Depends(get_db)):
return db.query(User).all()
@app.get('/users/{uid}', response_model=UserOut)
def read(uid: int, db: Session = Depends(get_db)):
u = db.query(User).filter(User.id == uid).first()
if not u: raise HTTPException(404, 'not found')
return u
@app.put('/users/{uid}', response_model=UserOut)
def update(uid: int, body: UserIn, db: Session = Depends(get_db)):
u = db.query(User).filter(User.id == uid).first()
if not u: raise HTTPException(404)
u.name = body.name; u.email = body.email
db.commit(); db.refresh(u); return u
@app.delete('/users/{uid}')
def delete(uid: int, db: Session = Depends(get_db)):
u = db.query(User).filter(User.id == uid).first()
if not u: raise HTTPException(404)
db.delete(u); db.commit(); return {'deleted': uid}
if __name__ == '__main__':
uvicorn.run(app, host='0.0.0.0', port=8000)
# Create a user
curl -X POST http://localhost:8000/users \
-H 'Content-Type: application/json' \
-d '{"name":"Ada","email":"ada@x.com"}'
# List all users
curl http://localhost:8000/users
# Get user 1
curl http://localhost:8000/users/1
# Update user 1
curl -X PUT http://localhost:8000/users/1 \
-H 'Content-Type: application/json' \
-d '{"name":"Ada L","email":"ada@y.com"}'
# Delete user 1
curl -X DELETE http://localhost:8000/users/1
How it works
CRUD = Create, Read, Update, Delete — the four operations every data app needs. The pattern: (1) SQLAlchemy model maps a Python class to a DB table. Column(Integer, primary_key=True) marks the id. (2) Pydantic schemas (UserIn for input, UserOut for output) validate and serialise. from_attributes=True lets Pydantic read from ORM objects. (3) @app.post creates, @app.get reads, @app.put updates, @app.delete deletes. (4) Depends(get_db) gives each request its own DB session and closes it automatically. (5) HTTPException(404) returns a proper 404 with a JSON error body. (6) response_model=UserOut means FastAPI validates the response against the schema before sending — leaks nothing extra. Test with curl, the Swagger UI at /docs, or fetch() from a browser.
⚡ Open in Notebook