Concepts 🚀 Project 🧮 Labs
📚 Learn / Mini Projects / CRUD API with FastAPI + SQLAlchemy

CRUD API with FastAPI + SQLAlchemy

Create, Read, Update, Delete users — the four operations every backend needs. SQLite + Pydantic + SQLAlchemy.

🚀 Mini Project · 2 files

Project files

📄 main.py python
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)
📄 test.sh bash
# 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
🔍 Where this is used in OpenBenchML OpenBenchML's models.py, datasets.py, competitions.py all follow this exact CRUD pattern with SQLAlchemy + Pydantic.