FastAPI(2): MySQL CRUD

잘 안되서 고생을 좀 했다.(아직 해결은 못한 부분이 있지만.. 정리 된거만 블로깅)

우선 코드스테이츠에서 토크세션을 진행했는데 "출판사가 OK하는 글쓰기" 를 주제로 최현우 프로 에디터님이 주관하여 세션을 진행하셨다.

간단히 요약하고 블로깅을 시작해보겠습니다.

글을 쓰면서 주의 할점

• 단문으로 글쓰기

  • 글은 길어지면 길어질수록 비문이 된다.
  • 아무리 길어도 한문장이 한줄 반 이상을 넘어가면 안된다.(비문이 될 확률이 높아진다)

• 첫문장에는 정의로 정어주면 좋다 ex) string은 문자열을 담는 자료형이다.

• 오늘 배운내용은 최대한 빨리 블로깅을 하는게 좋다.(나중에 블로깅하면 내가 왜 이렇게 코드를 짯는지 잘 모를수 있다.)

• 번역투로 글쓰기

FastAPI <-> MySQL 통신 전 준비사항

모듈 설치

poetry add fastapi uvicorn sqlalchemy pymysql async-generator async-exit-stack

sqlalchemy는 파이썬의 ORM 라이브러리의 하나이며, 관계형 데이터베이스에 접근하기 위한 레퍼이며 오브젝트 지향형으로 데이터베이스 access를 간단하게 접근이 가능함.

async-generator, async-exit-stack는 공식문서에서 python3.7이상 3.6인경우는 설치하라고 적혀있었고, 설치를 안하니 나중에 서버를 돌릴때 에러가 발생함.

공식문서 보며 따라하기

FastAPI는 레퍼런스는 많지는 않지만 , 공식문서가 엄청나게 자세하게 되어있어 좋은듯 하다.

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

공식문서 대로 파일을 이렇게 만들었고, __init.py 폴더의 내용은 비었있지만 sql_app이 python파일이 있는 패키지라고 알려주는 파일입니다.(비어있다고 삭제하면 x)

1. database.py 작성(database 연결과 관련된 로직)

공식문서에서는 sqllite와 postgresql이 예시로 되어있지만 postgresql 예시와 비슷하게 하면 동작은 된다.

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


SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:1234@localhost:3306/test_db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL
)


SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

• SQLALCHEMY_DATABASE_URL은 mysql을 사용할 경우 "mysql+pymysql://root:1234@localhost:3306/test_db" 이렇게 작성하면 되고

  • root: username /1234: 데이터베이스 비밀번호 / localhost: hostname /
    3306: port test_db: 스키마 이름

  • 실제 사용시에는 환경변수로 설정(보안목적)

• engin변수는 sqlalchemy engine을 만드는것이며 나중에 main.py 폴더에서 사용할 예정

• SessionLocal의 역할은 잘 모르겠지만 SessionLocal 클래스의 인스턴스를 생성하면 실제 데이터베이스 접근해서 CRUD를 할 수 있는거 같음.

• Base는 나중에 이 클래스에서 상속하여 각 데이터베이스 모델 또는 클래스 (ORM 모델)를 만듭니다.

---

2. 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(20), unique=True, index=True)
    hashed_password = Column(String(100))
    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(20), index=True)
    description = Column(String(20), index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

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

테이블 관계에 대해서는 나중에 블로깅을 할 예정이고(아직 잘몰라서 ..) Item에 owner_id가 있는걸 보아 1(User):N(Item)관계인걸 알 수 있다.

---

3.schemas.py 작성(타입설정)

from typing import List, Optional

from pydantic import BaseModel # 객체 타입설정


class ItemBase(BaseModel):
    title: str
    description: Optional[str] = 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


    class Config:
        orm_mode = True

schemas.py는 테이블의 타입을 설정 해놓은 파일

---

4.crud.py 파일작성(데이터베이스의 데이터와 상호작용 및 재사용함수가 있음)

공식문서의 crud.py에서 create_user 코드만 가져와서 살펴보기

from sqlalchemy.orm import Session
from . import models, schemas
import sql_app.database


# import sql_app.models # 절대경로로 불러오는방법
# import sql_app.schemas # 절대경로로 불러오는방법

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
    

파라미터로 받은 db를 활용해서 실제 쿼리를 날리고 user는 페이로드값이라고 보면된다.
(실제 함수 실행부분은 main.py에서 진행) user의 타입은 schemas.UserCreate라고 적혀있는데
schemas.py폴더에 있는 UserCreate를 가져와서 타입핑(타입힌팅)을 한거라고 보면된다.
(email: str, password: str 이렇게 설정이 되어 있음)

db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
models에 있는 User테이블을 불러와서 (제일 최상단에 임포트가 되어 있음) 바디값을 넣어주면된다.

(email(실제 테이블컬럼명) = 데이터, hashed_password= 데이터)

create

데이터를 insert 하는 과정을 보면 db.add(db_user => 인스턴스 변수를 넣어주고)
db.commit()을 해야 데이터베이스에 적용이 된다. db.refresh(db_user) 마지막으로 이렇게 해줘야지 인스턴스변수에 변경된 내용이 적용된다.

여기서 주의할 점은 패스워드 해싱화 할때는 공식문서에 내용처럼 하면 안되고 암호화 모듈을 사용하던지 자신이 직접 암호화를 해야된다.

update

email_update = db.query(models.User).filter_by(email = 'kim').first()

email_update.email = 'hyunjin'

db.add(email_update)

db.commit()

delete

email_delete = db.query(models.User).filter_by(email = 'hyunjon').first()

db.delete(email_delete)

db.commit()

read 및 간단 ORM 연습

• db.query(models.User).all() User 테이블 모든 데이터 가져오기

  # for문을 돌려서 data확인
  
  for data in db.query(models.User).all():
    print(data.email, data.password)
  

• db.query(models.User).filter(models.User.email == 'email') 필터링

  • 주의점은 filter안에 조건식을 models.User.email 이렇게 가져와야됨
  • filter(models.User.email == 'email', models.User.password == '1234') 여러가지 조건을 넣어도 된다.
  • 하나만 가져올때는 제일뒤에 first()를 적어주면됨 db.query(models.User).filter(models.User.email == 'email').first()

• db.query(models.User).filter_by(email='kims', hashed_password="a").first()

  • filter와 같으나 필터링 할때 테이블명으로 조회가 가능함.

• db.query(models.User).limit(10).all() limit 10개만 가져오기

• db.query(models.User).offset(5).limit(10).all() => 6번부터 10개 가져오기(6,7,8,9,10,11,12,13,14,15)

• db.query(models.User).order_by(models.User.id.desc()) # 내림차순

여기까지만 하고 나중에 정리

main.py 작성

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
from sqlalchemy.orm import Session, sessionmaker

import sqlalchemy.orm.session

import sql_app.models
import sql_app.database
import sql_app.schemas
import sql_app.crud


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

app = FastAPI()


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

@app.post("/users",response_model=sql_app.schemas.User)
def create_user2(user:sql_app.schemas.UserCreate, db: Session = Depends(get_db)): # 무조건 typing을 해줘야 에러가 발생하지 않음
    db_user = sql_app.crud.get_user_by_email(db, email=user.email)

    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return sql_app.crud.create_user(db=db, user=user)

일부분만 공식문서에서 가져와서 연습을 해보았다. main.py 설명은 내일로.....

TODO:

sqlalchemy 관계설정 방법 보기

Alembic 마이그레이션 방법 알아보기 (오늘 이게 잘안되서 고생 ㅠ)

GO lang 기본문법 공부