FastAPI 공부 3일차

1. 주요 구성요소

from fastapi import FastAPI
import uvicorn

# 핵심 인스턴스로, 모든 라우팅, 미들웨어, 이벤트핸들러 등을 관리한다.
app = FastAPI()

if __name__ == '__main__':
  uvicorn.run(app, host='localhost', port=8000)

 

•  C나 Java같은 프로그래밍 언어에선 항상 main()이라는 함수를 시작으로 프로그램을 실행시키는데, 파이썬은 main함수가 존재하지 않고 들여쓰기를 통해 코드 실행의 레벨을 결정한다.
• __name__ 내장변수는 현재 모듈의 이름을 담고있는 내장변수로, 직접 실행된 모듈의 경우 __main__이라는 값을 갖게되며, 직접 실행하지 않은 import된 모듈은 모듈의 이름(파일명)을 갖게 된다. 결론은 아래와 같다.

 

if문은 Python에서 직접 실행된 스크립트인지 임포트된 모듈인지 구분하는 용도로 사용되는 조건문이다.
즉, 현재 스크립트 파일이 시작점이 맞는지 판단하는 작업이다.

 

위의 방식을 사용하면 python main.py로 서버를 실행할 수 있는데, main파일에 uvicorn을 임포트하지 않고 uvicorn main:app --reload 명령어로 서버를 실행할 수 있다. 개발단계에서는 --reload 명령어를 사용해 저장하면 코드 변경을 감지하여 서버를 자동으로 재시작하는 방법을 사용할 수 있어 편리하게 사용된다. 운영단계에선 주로 docker 이미지로 사용하기 때문에 나는 개발단계에선 if문을 사용하지 않고, --reload 옵션을 사용했다.

 

2. 파일 구조

• routers : 각종 API 경로를 정의. API Router를 사용하여 특정 리소스의 모든 라우트를 그룹화하여 main.py에 연결
• models : db 스키마 모델을 정의. SQLAlchemy와 같은 ORM 라이브러리를 사용. database폴더 안에 orm.py로 정의해도 됨.
• services : 비즈니스로직을 처리. 복잡한 연산, 외부 API 호출 등의 작업을 수행
• repositoris : 데이터 접근 로직을 처리. db와의 상호작용에 관련된 작업을 수행하며, SQLAlchemy의 메서드가 주로 사용됨.
• schemas : 입력 데이터 검증 및 데이터의 직렬화를 담당하는 Pydantic 모델을 정의. request와 response로 나누어 작성함.

# schema/request.py 파일 예시. Models 폴더로 만들기도 함.
from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
  contents: str
  is_done: bool

class SignUpRequest(BaseModel):
  username: str
  password: str

class LogInRequest(BaseModel):
  username: str
  password: str

 

Pydantic 코드 작성순서.
1. pydantic의 BaseModel을 상속받은 class로 데이터 타입을 정의
2. @app.post() 데코레이터를 사용해 post 메소드의 함수 생성
3. 함수의  인자로 pydantic으로 만든 타입을 받음

 

# api/todo.py
@router.post("", status_code=201)
def create_todo_handler(
  request: CreateToDoRequest,
  todo_repo: ToDoRepository = Depends(),
  ) -> ToDoSchema:
  todo: ToDo = ToDo.create(request=request) # id=None
  todo: ToDo = todo_repo.create_todo(todo=todo) # id=int
  return ToDoSchema.from_orm(todo)

 

이 api 엔드포인트 코드를 보면 request의 body를 CreateToDoRequest로 파싱하고 있다. 그리고 리턴값에 대한 타입으로는 ToDoSchema를 받는데, 이 타입이 response에 정의된다. 

# schema/response.py
from typing import List
from pydantic import BaseModel


class ToDoSchema(BaseModel):
  id: int
  contents: str
  is_done: bool
# orm 객체를 pydantic 객체로 변환 가능
  class Config:
    orm_mode = True

class ToDoListSchiema(BaseModel):
  todos: List[ToDoSchema]

 

ToDo의 응답 스키마도 역시 BaseModel을 상속받지만, orm_mode를 추가하여 SQLAlchemy orm 객체를 pydantic 모델로 변환할 수 있도록 허용한다. 이 설정 덕분에 .from_orm() 메서드를 사용해 ORM 객체(ToDo)를 직접 받아 Pydantic 인스턴스를 생성할 수 있다. 이 변환 과정이 바로 직렬화에 해당하며,  from_orm 메서드가 변환한 Pydantic 모델을 FastAPI가 JSON으로 변환해 클라이언트에 응답한다.

 

# database/orm.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import declarative_base, relationship
from schema.request import CreateToDoRequest

# 이 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"))

  def __repr__(self):
    return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
  
  @classmethod
  def create(cls, request: CreateToDoRequest) -> "ToDo":
    return cls(
      contents=request.contents,
      is_done=request.is_done
    )

 

@classmethod 덕분에 ToDo.create()의 형태로 사용할 수 있다. __repr__()은 객체의 공식적인 문자열 표현을 반환하는데, orm 클래스에서 주로 사용되어 디버깅, 로깅, 콘솔 출력에 사용되며 정확하고 객체 복원가능한 표현으로 볼 수 있다. create 함수는 request에서 받은 Pydantic 객체를 기반으로 ToDo 객체를 생성한다. 이 함수가 사용될 시점엔 db에 저장되지 않은 상태로 메모리상 객체의 형태이다. 

 

# repository/todo.py
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:
# Depends를 통해 get_db를 의존성 주입받아 사용
  def __init__(self, session: Session = Depends(get_db)):
    self.session = session

  def get_todos(self) -> List[ToDo]:
  # 결괏값에서 단일 orm 객체들만 추출 -> list()로 ScalarResult를 Python list로 변환
    return list(self.session.scalars(select(ToDo)))

  def get_todo_by_todo_id(self, todo_id: int) -> ToDo | None:
    return self.session.scalar(select(ToDo).where(ToDo.id == todo_id))

  def create_todo(self, todo: ToDo) -> ToDo:
  # orm 객체를 db 트랜잭션에 추가(Insert 대기)
    self.session.add(instance=todo)
  # 실제로 INSERT 쿼리를 날려 저장
    self.session.commit()  # db save
  # db에서 새로 읽어와 id 등 자동생성 필드를 채움
    self.session.refresh(instance=todo)  # db read -> todo의 id값이 결정됨.
    return todo

 

다음으로는 db에 저장하기 위해 레포지토리 패턴을 사용했다. 데이터 접근을 위한 계층으로 Depends를 통해 get_db를 의존성 주입받아 사용했다. sqlalchemy에서 session은 db와의 연결을 관리하고 orm 객체를 추적하며, 트랜잭션을 시작/커밋/롤백하는 도구이다. FastAPI에서 주입받은 DB 세션 인스턴스로 HTTP 요청 1건마다 새로운 세션이 만들어지고, 요청 종료 시 자동으로 종료된다.

 

- 실행흐름

쿼리 실행 관련 메서드
- excute() : SQL 또는 ORM 기반 쿼리를 실행. law sql, 복잡한 select/join에 사용
- scalar() : 쿼리 결과의 첫번째 컬럼값 하나만 반환. ID나 count 등 단일 값 조회
- add(obj) : 새 객체를 세션에 추가(INSERT 준비). 객체 생성(create)에 사용
- commit() : 현재까지의 변경사항을 DB에 저장하고 트랜잭션 종료.
- rollback() : 트랜잭션 내 모든 변경사항을 취소하고 초기화. 에러발생시 사용
- refresh(obj) : db에서 최신 상태로 재조회하여 객채의 필드를 갱신. commit() 후 auto-generate필드 반환
- expire(obj) : 해당 객체를 만료 상태로 만들어 다음에 접근할 때 db에서 다시 조회하게 함
- fluch() : db에 즉시 반영. 트랜잭션은 유지됨(commit은 x). INSERT/UPDATE를 즉시 반영해야 할 때