[FastAPI] database

본 포스팅은 인프런에서 제공하는 '실전! FastAPI 입문' 강의를 수강후에 정리 및 복습을 위해 작성하는 글입니다.

코드 분석

---

database > connection.py

• SQLAlchemy를 사용하여 데이터베이스와 연결을 설정하는 파일. 데이터베이스의 접속정보와 접근할 데이터베이스의 정보를 명시하고 있음.
• 데이터베이스 세션을 생성하고, 커밋 여부를 설정.
• 이후 API에서 데이터베이스와의 연결이 필요할 경우 생성된 SessionFactory를 통해 데이터를 전송 또는 제공.

코드 분석

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

#데이터베이스 접속정보를 명시
DATABASE_URL = "mysql+pymysql://root:todos@127.0.0.1:3306/todos"

#echo=True 쿼리 실행 시 해당 쿼리 print해준다. 디버깅용.

#데이터베이스와 상호작용을 관리하는 엔진 생성
engine = create_engine(DATABASE_URL)

#SQLAlchemy의 sessionmaker를 통해 데이터베이스 세션을 생성하는 팩토리 생성
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

#generator로 의존성 주입을 사용할 때, 데이터베이스 세션을 관리하는 함수
def get_db():
    session = SessionFactory()
    try:
    	#yield를 사용하면 세션을 반환한 후에도 세션을 관리할 수 있음.
        yield session
    finally:
    	#yield 이후 세션을 종료
        session.close()

 

database > orm.py

• SQLAlchemy를 통해 ORM을 구현
  *ORM : Python의 객체와 데이터베이스의 테이블 간 매핑을 처리
  쿼리를 직접 작성하지 않고도 CRUD 가능.
• 데이터 모델을 정의하고 create()와 같이 객체를 생성하는 매서드를 내장할 수 있음.

코드 분석

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

from schema.request import CreateToDoRequest

#declarative_base() : 데이터베이스 모델의 기본 클래스를 생성하는 함수.
Base = declarative_base()

class ToDo(Base):
    #데이터베이스 테이블 이름
	__tablename__ = "todo"

	#컬럼
    id = Column(Integer, primary_key=True, index=True)
    contents = Column(String(256), nullable=False)
    is_done = Column(Boolean, nullable=False)
    user_id = Column(Integer, ForeignKey("user.id"))

	#__repr__ : 객체를 문자열로 표현할 때 어떻게 나타낼지를 정의하는 메서드
    def __repr__(self):
        return f"Todo(id={self.id}, contents={self.contents}, is_done={self.is_done})"

    @classmethod #일반적인 매서드와 달리 class를 인자로 받는 매서드
    def create(cls, request: CreateToDoRequest):
        return cls(
            contents= request.contents,
            is_done= request.is_done,
        )

    def done(self) -> "ToDo":
        self.is_done = True
        #send email과 같이 인스턴스 메서드에 작성해놓으면 추가 기능 구현에 유리
        return self

    def undone(self) -> "ToDo":
        self.is_done = False
        return self

class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(256), nullable=False)
    password = Column(String(256), nullable=False)
    # relationship으로 User와 Toodo의 관계를 명시해두면
    # scalar로 조회했을 때 두 테이블이 조인되어 결과가 나옴
    todos = relationship("ToDo", lazy="joined")

    @classmethod
    def create(cls, username: str, hashed_password: str) -> "User":
        return cls(
            username=username,
            password=hashed_password,
        )

 

# ORM의 데이터 조회 방법
# Lazy Loading
#   지연 로딩 : 연관된 객체의 데이터가 실제 필요한 시점에 조회
#   장점 : 첫 조회 속도가 더 빠름
#   단점 : N+1 문제 발생
# N+1 문제란?
# 데이터의 갯수만큼 조인이 발생함.
#   for td in todos:
#       print(td.user.username)

# Eager Loading
#   즉시 로딩 : 데이터를 조회할 때 처음부터 연관된 객체를 같이 읽어옴
#   장점 : 데이터를 더 효율적으로 가져올 수 있음 (N+1 발생X)
#   단점 : 꼭 필요하지 않은 데이터까지 JOIN하여 읽어올 수 있음.

 

database > repository.py

• SQLAlchemy의 ORM을 사용하여 데이터베이스와 상호작용을 담당. Spring MVC에서의 Repository와 동일한 역할

from typing import List

from fastapi import Depends
from sqlalchemy import select, delete
from sqlalchemy.orm import Session

from database.connection import get_db
from database.orm import ToDo, User


class ToDoRepository:
	#connection의 get_db매서드를 통해 의존성을 주입받아 세션을 엶
    def __init__(self, session: Session = Depends(get_db)):
        self.session = session

    def get_todos(self) -> List[ToDo]:
		#self.session.scalars(select())를 통해 select 쿼리를 수행하고 결과를 list형태로 받아옴
        return list(self.session.scalars(select(ToDo)))

    def get_todo_by_todo_id(self, todo_id: int) -> ToDo | None:
    	#self.session.scalars(select().where)를 통해 where절을 포함한 select쿼리를 수행
        return self.session.scalar(select(ToDo).where(ToDo.id == todo_id))

	#orm에서 가져온 ToDo객체를 세션에 담아 insert쿼리를 수행, todo객체를 다시 반환
    def create_todo(self, todo: ToDo) -> ToDo:
        self.session.add(instance=todo)
        self.session.commit() #db 저장
        self.session.refresh(instance=todo) #db에서 todo_id를 생성, 다시 To_Do에 담아 return
        return todo

	#orm에서 가져온 ToDo객체를 세션에 담아 update쿼리를 수행, todo객체를 다시 반환
    def update_todo(self, todo: ToDo) -> ToDo:
        self.session.add(instance=todo)
        self.session.commit() #db 저장
        self.session.refresh(instance=todo)
        return todo

	#파라미터로 받은 todo_id를 세션에 담아 delete쿼리를 수행
    def delete_todo(self, todo_id: int) -> None:
        self.session.execute(delete(ToDo).where(ToDo.id == todo_id))
        self.session.commit()