RDB 연동 - 3

RDB 연동 - 1

RDB 연동 - 1

RDB 연동 - 2

RDB 연동 - 2

 

위의 RDB 연동 -1, RDB 연동 -2에 이어서 3번째 글을 작성해 보겠습니다.

 

우선 살펴볼 코드는 main.py의 전체 코드는 내용입니다.

 

import uvicorn
from typing import List

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

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

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


app = FastAPI()


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)):
    existed_user = db.query(models.User).filter_by(
        email=user.email
    ).first()

    if existed_user:
        raise HTTPException(status_code=400, detail="Email already registered")

    user = models.User(email=user.email, password=user.password)
    db.add(user)
    db.commit()

    return user


@app.get("/users", response_model=List[schemas.User])
def read_users(db: Session = Depends(get_db)):
    return db.query(models.User).all()

if __name__ == '__main__':
	uvicorn.run('main:app', reload = True)

 

위에서 from . import models, schemas를 사용해서 RDB 연동 - 2에서 만든 models와 schmas를 불러옵니다. 그리고 database에서 SessionLocal과 engine을 불러옵니다.

 

그 다음으로는 models.Base.metadata.create_all(bind=engine) 구문이 있습니다. 해당 코드의 의미는 우리가 Models에서 정의한 Table 즉, User Table을 DB에 만든다는 의미입니다.

 

다음 코드 부분을 보면, 

app = FastAPI()


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

 

이렇게 작성되어져 있습니다. 우선 FastAPI 앱을 만들고, DB를 사용하기 위한 helper 함수인 get_db를 정의했습니다.

SessionLocal을 사용해서 db라는 Session을 생성을 하고, db를 yeield를 해줍니다.

 

yeield의 경우, 반환하고 함수의 역할일 끝나는 return과는 달리 yeield는  반환하고 함수의 역할일 끝나지 않는 일종의 서브 루틴의 개념입니다.  

 

어떤 error가 나든 반드시 db Session을 종료하도록 설정되어져 있습니다.

 

다음 코드 부분을 보면, 이렇게 작성되어져 있습니다.

@app.post("/users", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    existed_user = db.query(models.User).filter_by(
        email=user.email
    ).first()

    if existed_user:
        raise HTTPException(status_code=400, detail="Email already registered")

    user = models.User(email=user.email, password=user.password)
    db.add(user)
    db.commit()

    return user

 

위의 API는 User를 생성하기 위한 API입니다. 위에 코드에서 create_user는 user와 db를 입력으로 받는데, 그중 schemas에 UserCreate를 입력으로 받습니다.

 

schemas에 UserCreate를 보면 다음과 같습니다.

from pydantic import BaseModel


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    email: str
    is_active: bool

    class Config:
        orm_mode = True

 

UserCreate는 UserBase를 상속 받습니다. 즉 UserCreate는 email과 password를 입력으로 받습니다.

 

db라는 parameter에 Depends 함수를 사용해서 get_db를 입력으로 받습니다. 그리고 아래 부분의 코드에서

 

existed_user = db.query(models.User).filter_by(
        email=user.email
    ).first()

 

db에 우리가 만든 Table인 User라는 곳에 우리가 입력 받은 user의 email의 값을 filter_by라는 함수를 통해서 db에 있는 User라는 Table에 email이 존재하는지 확인을 하는 코드입니다.

 

if existed_user:
        raise HTTPException(status_code=400, detail="Email already registered")

 

그리고 이 코드와 같이 만약 email이 이미 있는 경우에는 HTTPException을 통해서 이메일이 이미 등록을 했다.라는 것을 반환하게 됩니다.

 

만약에 없는 경우에, 다음 코드를 사용해서

user = models.User(email=user.email, password=user.password)

 

email과 password를 입력을 가지고 있는 user table을 만들고, db에 추가하고 commit을 해주어 db에 반영을 해주게 됩니다. 위와 같은 방법을 사용해서 user를 간단하게 생성을 했습니다.

 

다음은 read_users에 해당하는 코드입니다. db에서 user를 읽고 반환을 합니다.

@app.get("/users", response_model=List[schemas.User])
def read_users(db: Session = Depends(get_db)):
    return db.query(models.User).all()

 

 

새성된 Session은 db에 query를 날리고 해당 대상은 User라는 Table입니다.

 

all이라는 함수를 사용해서 User에 있는 모든 기록들을 가져 오게됩니다.