fastapi_sqlalchemy_sqlite_crud_user_api_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)
16
17Base = declarative_base()
18
19# 2. Database Models (SQLAlchemy)
20class User(Base):
21    __tablename__ = "users"
22
23    id = Column(Integer, primary_key=True, index=True)
24    email = Column(String, unique=True, index=True)
25    hashed_password = Column(String)
26
27# Create the database tables
28Base.metadata.create_all(bind=engine)
29
30# 3. Pydantic Models (Schemas)
31class UserBase(BaseModel):
32    email: str
33
34class UserCreate(UserBase):
35    password: str
36
37class UserOut(UserBase):
38    id: int
39
40    class Config:
41        from_attributes = True
42
43# 4. Dependency
44def get_db():
45    db = SessionLocal()
46    try:
47        yield db
48    finally:
49        db.close()
50
51# 5. FastAPI App and Routes
52app = FastAPI()
53
54@app.post("/users/", response_model=UserOut)
55def create_user(user: UserCreate, db: Session = Depends(get_db)):
56    db_user = db.query(User).filter(User.email == user.email).first()
57    if db_user:
58        raise HTTPException(status_code=400, detail="Email already registered")
59    
60    # In a real app, you would hash the password
61    fake_hashed_password = user.password + "notreallyhashed"
62    new_user = User(email=user.email, hashed_password=fake_hashed_password)
63    
64    db.add(new_user)
65    db.commit()
66    db.refresh(new_user)
67    return new_user
68
69@app.get("/users/", response_model=List[UserOut])
70def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
71    users = db.query(User).offset(skip).limit(limit).all()
72    return users