[FastAPI] FastAPI [25] SQL (Relational) Databases (2)

https://fastapi.tiangolo.com/tutorial/sql-databases/

SQL (Relational) Databases - FastAPI

<FastAPI 공식문서 참조>

 

* 공식문서 순서상으론 Dependencies(의존성), Security(보안) 다음 SQL Databases이나 필요에 의해 본 Chapter부터 포스팅합니다.

 

 

 

0. 전체 디렉토리 구조

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

 

1. Main FastAPI app

 sql_app/main.py 파일을 보자. 이전에 만든 모든 다른부분을 사용해 여기다 합칠 것이다.

 

전체 구성은 다음과 같다.

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

 

1-1. Create the database tables

models.Base.metadata.create_all(bind=engine)

 위 코드를 통해 아주 간단하게 데이터베이스 테이블을 만들 수 있다.

 

1-1-1. Alembic Note

보통 내 데이터베이스의 시작 설정을 Alembic을 통해 사용해야 한다.

이는 migrations를 위해 보통 사용된다(이게 주된 작업이다.)

 

migration이란, SQLAlchemy 모델에서 구조변경, 새 속성 추가 시 등에 DB에서 이런 변경사항을 복제해 구현할 때마다 필요한 단계를 의미한다.

 

다시, Alembic이란, Python으로 SQLAlchemy를 사용하고 있을 때 DB를 관리해주는 migration tool이다.

 

1-2. Create a dependency

 sql_app/database.py에 생성했던 SessionLocal 클래스를 사용해 현재 파일에 의존성을 생성할 것이다.

 

매 요청마다 독립적인 데이터베이스 세션/연결이 필요하며, 모든 요청에 대해 동일한 세션을 사용한 다음, 요청이 완료되면 닫아야 한다.

 

그러고 나서 다음 요청을 위해 새로운 세션이 만들어져야 한다.

 

이를 위해서 yield를 사용해 새로운 의존성을 만들 것이다. 이는 의존성 부분에 설명되어있다. https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/

 

이 의존성은 단일 요청에 사용될 것이며 이 요청이 완료되면 닫히는 새로운 SQLAlchemy의 SessionLocal을 만들 것이다.

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

 

 

다음 예제를 보면, path 연산 함수에 의존성을 사용할 때는 SQLAlchemy에서 직접 import한 Session 타입의 변수를 직접 생성하여 사용했다.

 

#post 연산함수
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):

#get 연산함수
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):

#post 연산함수
def create_item_for_user(user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):

#get 연산함수
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):

 

 

 

2. Create your FastAPI path operations

 이 파트는 표준 FastAPI 경로 연산 코드이다.

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

 매 세션에 관한 의존성과 path operation function에 설정한 의존성 덕분에,

함수 내부에서 crud.get_user을 간단히 바로 호출할 수 있다.

 

 

3. About  def  vs  async def 

SQLAlchemy 자료에는 await 키워드를 직접 사용할 수는 없다.

 

아래와 같은 코드는 불가능하다.

user = await db.query(User).first()

 

그 대신 아래와 같이 쓴다.

user = db.query(User).first()

 

그래서 def 키워드를 async def라 사용하지 않고 그냥 def로 사용한다.

 

 

 

4. Migrations

 'Alembic 하나만'을 사용하여 직접 migrations 설정이 가능하다.

이는 FastAPI와 연관된 라이브러리가 아니기 때문이다. 따라서 FastAPI로 구현되지 않은 부분도 동일한 SQLAlchemy 모델 및 유틸리티를 사용할 수 있다.

 

 

5. Review all the files

 가장 상위 디렉토리의 이름이  my_super_project  이며 서브디렉토리의 이름을  sql_app  으로 설정했다는 것을 기억하자.

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

sql_app 파일은 init.py 파일과 database를 무조건 가지고 있어야 한다.

 

• sql_app/__init__.py 는 비어있는 파일이다.
• sql_app/database.py는 다음과 같다.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

 

• sql_app/models.py

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

 

• sql_app/schemas.py:

 

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

 

• sql_app/crud.py:

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

 

• sql_app/main.py:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

 

 

6. Interact with the database directly

https://sqlitebrowser.org/

DB Browser for SQLite

여기서 데이터베이스와 직접 상호작용할 수도 있다.

 

 

 

7. Reference

https://fastapi.tiangolo.com/tutorial/sql-databases/

SQL (Relational) Databases - FastAPI