fastapi_sqlalchemy_sqlite_user_crud_quickstart.py

1from typing import List
2
3from fastapi import Depends, FastAPI, HTTPException
4from sqlalchemy import create_engine, Column, Integer, String
5from sqlalchemy.ext.declarative import declarative_base
6from sqlalchemy.orm import sessionmaker, Session
7from pydantic import BaseModel
8
9# 1. Database Configuration
10SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
11
12engine = create_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# 2. 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# Create the database tables
26Base.metadata.create_all(bind=engine)
27
28# 3. Pydantic Schemas
29class UserBase(BaseModel):
30    email: str
31
32class UserCreate(UserBase):
33    password: str
34
35class UserResponse(UserBase):
36    id: int
37
38    class Config:
39        from_attributes = True
40
41# 4. Dependency to get DB session
42def get_db():
43    db = SessionLocal()
44    try:
45        yield db
46    finally:
47        db.close()
48
49# 5. FastAPI Application
50app = FastAPI()
51
52@app.post("/users/", response_model=UserResponse)
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    
62    db.add(new_user)
63    db.commit()
64    db.refresh(new_user)
65    return new_user
66
67@app.get("/users/", response_model=List[UserResponse])
68def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
69    users = db.query(User).offset(skip).limit(limit).all()
70    return users
71
72@app.get("/users/{user_id}", response_model=UserResponse)
73def read_user(user_id: int, db: Session = Depends(get_db)):
74    db_user = db.query(User).filter(User.id == user_id).first()
75    if db_user is None:
76        raise HTTPException(status_code=404, detail="User not found")
77    return db_user