fastapi_crud_users_items_sqlalchemy_sqlite.py

1from fastapi import Depends, FastAPI, HTTPException
2from sqlalchemy import Column, ForeignKey, Integer, String, create_engine
3from sqlalchemy.ext.declarative import declarative_base
4from sqlalchemy.orm import Session, sessionmaker, relationship
5from pydantic import BaseModel
6from typing import List, Optional
7
8# --- DATABASE SETUP ---
9SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
10engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
11SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
12Base = declarative_base()
13
14# --- MODELS (SQLAlchemy) ---
15class DBUser(Base):
16    __tablename__ = "users"
17    id = Column(Integer, primary_key=True, index=True)
18    email = Column(String, unique=True, index=True)
19    items = relationship("DBItem", back_populates="owner")
20
21class DBItem(Base):
22    __tablename__ = "items"
23    id = Column(Integer, primary_key=True, index=True)
24    title = Column(String, index=True)
25    owner_id = Column(Integer, ForeignKey("users.id"))
26    owner = relationship("DBUser", back_populates="items")
27
28Base.metadata.create_all(bind=engine)
29
30# --- SCHEMAS (Pydantic) ---
31class ItemCreate(BaseModel):
32    title: str
33
34class Item(ItemCreate):
35    id: int
36    owner_id: int
37    class Config:
38        orm_mode = True
39
40class UserCreate(BaseModel):
41    email: str
42
43class User(UserCreate):
44    id: int
45    items: List[Item] = []
46    class Config:
47        orm_mode = True
48
49# --- DEPENDENCY ---
50def get_db():
51    db = SessionLocal()
52    try:
53        yield db
54    finally:
55        db.close()
56
57# --- APP AND ROUTES ---
58app = FastAPI()
59
60@app.post("/users/", response_model=User)
61def create_user(user: UserCreate, db: Session = Depends(get_db)):
62    db_user = DBUser(email=user.email)
63    db.add(db_user)
64    db.commit()
65    db.refresh(db_user)
66    return db_user
67
68@app.get("/users/", response_model=List[User])
69def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
70    return db.query(DBUser).offset(skip).limit(limit).all()
71
72@app.get("/users/{user_id}", response_model=User)
73def read_user(user_id: int, db: Session = Depends(get_db)):
74    db_user = db.query(DBUser).filter(DBUser.id == user_id).first()
75    if db_user is None:
76        raise HTTPException(status_code=404, detail="User not found")
77    return db_user
78
79@app.post("/users/{user_id}/items/", response_model=Item)
80def create_item_for_user(user_id: int, item: ItemCreate, db: Session = Depends(get_db)):
81    db_item = DBItem(**item.dict(), owner_id=user_id)
82    db.add(db_item)
83    db.commit()
84    db.refresh(db_item)
85    return db_item
86
87@app.get("/items/", response_model=List[Item])
88def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
89    return db.query(DBItem).offset(skip).limit(limit).all()