fastapi_crud_api_with_sqlalchemy_pydantic_sqlite.py

1from typing import List, Optional
2
3from fastapi import Depends, FastAPI, HTTPException
4from sqlalchemy import Column, ForeignKey, Integer, String, create_database_engine
5from sqlalchemy.ext.declarative import declarative_base
6from sqlalchemy.orm import Session, sessionmaker
7from pydantic import BaseModel
8
9# --- DATABASE SETUP ---
10SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
11
12engine = create_database_engine(
13    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
14)
15SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
16Base = declarative_base()
17
18# --- SQLALCHEMY MODELS ---
19class User(Base):
20    __tablename__ = "users"
21    id = Column(Integer, primary_key=True, index=True)
22    email = Column(String, unique=True, index=True)
23    hashed_password = Column(String)
24
25# --- PYDANTIC SCHEMAS ---
26class UserBase(BaseModel):
27    email: str
28
29class UserCreate(UserBase):
30    password: str
31
32class UserSchema(UserBase):
33    id: int
34
35    class Config:
36        from_attributes = True
37
38# Create the database tables
39Base.metadata.create_all(bind=engine)
40
41app = FastAPI()
42
43# Dependency to get DB session
44def get_db():
45    db = SessionLocal()
46    try:
47        yield db
48    finally:
49        db.close()
50
51# --- CRUD OPERATIONS ---
52@app.post("/users/", response_model=UserSchema)
53def create_user(user: UserCreate, db: Session = Depends(get_db)):
54    db_user = db.query(User).filter(User.email == user.email).first()
55    if db_user:
56        raise HTTPException(status_code=400, detail="Email already registered")
57    
58    # In a real app, you would hash the password
59    fake_hashed_password = user.password + "notreallyhashed"
60    new_user = User(email=user.email, hashed_password=fake_hashed_password)
61    db.add(new_user)
62    db.commit()
63    db.refresh(new_user)
64    return new_user
65
66@app.get("/users/", response_model=List[UserSchema])
67def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
68    users = db.query(User).offset(skip).limit(limit).all()
69    return users
70
71@app.get("/users/{user_id}", response_model=UserSchema)
72def read_user(user_id: int, db: Session = Depends(get_db)):
73    db_user = db.query(User).filter(User.id == user_id).first()
74    if db_user is None:
75        raise HTTPException(status_code=404, detail="User not found")
76    return db_user
77
78@app.delete("/users/{user_id}")
79def delete_user(user_id: int, db: Session = Depends(get_db)):
80    db_user = db.query(User).filter(User.id == user_id).first()
81    if db_user is None:
82        raise HTTPException(status_code=404, detail="User not found")
83    db.delete(db_user)
84    db.commit()
85    return {"detail": "User deleted"}