Pydantic을 통해 받은 데이터를 ORM으로 변경 후 그 ORM을 통해 데이터 저장하는 실습 진행
01 ToDo 생성하기
1. orm.py 파일에 쓰기
Class ToDo(Base): 아래 클래스 메소드를 통해 요청 받은, 즉, request body를 전달 받은 부분을 ORM 개체로 변환해주는 클래스 메소드 만들기
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
from schema.request import CreateToDoRequest
Base = declarative_base()
# ToDo 클래스 모델링 한 것
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)
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(
# id값은 DB에 의해 자동으로 결정 되서 별도로 지정안해줌
contents=request.contents,
is_done=request.is_done,
)
2. main.py에 쓰기
ORM 객체를 생성하는 부분 추가하기
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
...
# # POST Medthod 사용하여 todo 생성 API
# @app.post("/todos", status_code=201)
# def create_todos_handler(request: CreateToDoRequest):
# todo_data[request.id] = request.dict()
# return todo_data[request.id]
# 위 내용을 아래와 같이 수정
# POST Medthod 사용하여 todo 생성 API
@app.post("/todos", status_code=201)
def create_todo_handler(
request: CreateToDoRequest,
session: Session = Depends(get_db),
) -> ToDoSchema:
todo: ToDo = ToDo.create(request=request) # id=None
todo: ToDo = create_todo(session=session, todo=todo) # id=int
return ToDoSchema.from_orm(todo)
3. repository.py에 쓰기
다음을 추가
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
...
def create_todo(session: Session, todo: ToDo) -> ToDo: # Session 및 Todo를 ORM 객체(session, todo)로 받기
session.add(instance=todo) # session에 todo 인스턴스 추가
session.commit() # DB에 저장
session.refresh(instance=todo) # DB에서 읽어오기 -> DB에 저장하면서 생성된 todo 객체에 todo_id값 또한 반영되어있음
return todo # 읽어온 todo 반환
4. request.py에 쓰기
DB에서 id를 직접 할당해주기 때문에 이제 더 이상 request body에서 id값을 받지 않아도 되므로 해당 파 수정: request.py에서 id 값 제거
request 스키마(request.py)와 response 스키마(response.py)를 분리시켜 놨기에 한 쪽 (예, request.py)에서만 컬럼을 변경해도 (i.e., id 컬럼 값을 지워도) 다른 쪽 (i.e., response.py)은 전혀 영향을 받지 않게
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용
...
class CreateToDoRequest(BaseModel):
# id: int # 삭제
contents: str
is_done: bool
5. response.py 수정
pydantic Version 2에서는 from_orm을 사용하기 위해선 orm_mode = True가 아닌 from_attributes = True라고 config를 설정해줘야됨
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List
class ToDoSchema(BaseModel):
id: int
contents: str
is_done: bool
class Config:
from_attributes = True # orm_mode = True에서 from_attributes = True로 수정
class ToDoListSchema(BaseModel):
todos: List[ToDoSchema]
6. 실습으로 확인 - SwaggerUI 문서에서
0) 실습 환경 준비
가상환경 활성화, docker 컨테이너 시작, 경로 src로 이동, uvicorn 실행
# 가상환경 활성화
$ . /c/Users/관리자/Desktop/projects/todos/Scripts/activate
# docker 컨테이너 목록 및 상태 확인
(todos)$ docker ps -a
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 2 weeks ago Exited (0) 24 hours ago todos
2597a124c27d docker/welcome-to-docker:latest "/docker-entrypoint.…" 2 weeks ago Exited (0) 2 weeks ago welcome-to-docker
# todos 컨테이너 시작하기
(todos)$ docker start todos
todos
# src 경로로 이동
(todos)$ cd /c/Users/관리자/Desktop/projects/todos/src
# uvicorn 실행
(todos)$ uvicorn main:app --reload
INFO: Will watch for changes in these directories: ['C:\\Users\\관리자\\Desktop\\projects\\todos\\src']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [8444] using WatchFiles
INFO: Started server process [6332]
INFO: Waiting for application startup.
INFO: Application startup complete.
1) 브라우저에 SwaggerUI 문서 띄우기
http://127.0.0.1:8000/docs 입력
POST /todos Create Todo Handler 클릭 > Try it out > Request body 아래 다음을 입력
{
"contents": "test",
"is_done": false
}
Execute 클릭
id 값을 주지 않아도 DB에서 자동으로 id 값이 매겨져서 출력됨 (i.e., "id": 11)
우리가 지정한대로 ("contents": "test")와 ("is_done": false)가 제대로 들어간 것을 확인
해당 데이터는 DB에 저장이 됐기에 서버를 내렸다 올려도 데이터가 유지가 됨
2) 데이터베이스에 저장됐는지 확인
서버 내리기: terminal에서 ctrl + C 누르기
다시 동작: CLI에서 # uvicorn main:app --reload 입력 후 Enter
venv 활성화 및 src 디렉토리로 이동 uvicorn 실행해야 됨
# cd /c/Users/관리자/Desktop/projects/todos/src
# . ../Scripts/activate
# uvicorn main:app --reload
Swagger UI 문서로 다시 이동: 브라우저에 http://127.0.0.1:8000/docs 입력
전체 ToDo 조회 해보기: GET /todos Get Todos Handler 클릭 > Try it out > Execute 클릭
방금 생성한 ToDo (id=11, contents="test", is_done=false)가 있는 것을 확인 (즉, DB에 저장된 것을 확인 )
이때까지의 코드들:
/c/Users/관리자/Desktop/projects/todos/src/main.py
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from typing import List
from database.connection import get_db
from database.repository import get_todos, get_todo_by_todo_id, create_todo
from database.orm import ToDo
from schema.response import ToDoSchema, ToDoListSchema
from schema.request import CreateToDoRequest
app = FastAPI()
# 첫 화면 API
@app.get("/")
def health_check_handler():
return {"ping": "pong"}
# 데이터베이스 역할하는 딕셔너리 생성
todo_data = {
1: {
"id": 1,
"content": "실전! FastAPI 섹션 0 수강",
"is_done": True,
},
2: {
"id": 2,
"content": "실전! FastAPI 섹션 1 수강",
"is_done": False,
},
3: {
"id": 3,
"content": "실전! FastAPI 섹션 2 수강",
"is_done": False,
},
}
# GET Method 사용하여 전체 조회 API - 딕셔너리 참조
# @app.get("/todos", status_code=200)
# def get_todos_handler(order: str | None = None):
# rt = list(todo_data.values())
# if order and order == "DESC":
# return rt[::-1]
# return rt
# GET Method 사용하여 전체 조회 API - DB 참조
@app.get("/todos", status_code=200)
def get_todos_handler(
order: str | None = None,
session: Session = Depends(get_db),
) -> ToDoListSchema:
todos: List[ToDo] = get_todos(session=session)
if order and order == "DESC":
# return todos[::-1]
return ToDoListSchema(
todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
)
# return todos
return ToDoListSchema(
todos=[ToDoSchema.from_orm(todo) for todo in todos]
)
# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(
todo_id: int,
session: Session=Depends(get_db),
) -> ToDoSchema:
todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
if todo:
return ToDoSchema.from_orm(todo)
raise HTTPException(status_code=404, detail="ToDo Not Fount")
# POST Medthod 사용하여 todo 생성 API
@app.post("/todos", status_code=201)
def create_todo_handler(
request: CreateToDoRequest,
session: Session = Depends(get_db),
) -> ToDoSchema:
todo: ToDo = ToDo.create(request=request) # id=None
todo: ToDo = create_todo(session=session, todo=todo) # id=int
return ToDoSchema.from_orm(todo)
# PATCH Method 사용하여 is_done 값 수정 API
@app.patch("/todos/{todo_id}", status_code=200)
def update_todo_handler(
todo_id: int,
is_done: bool = Body(..., embed=True)
):
todo = todo_data.get(todo_id)
if todo:
todo["is_done"] = is_done
return todo
raise HTTPException(status_code=404, detail="ToDo Not Found")
# DELETE Method 사용하여 todo 아이템 삭제 API
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo_handler(todo_id: int):
todo = todo_data.pop(todo_id, None)
if todo:
return
raise HTTPException(status_code=404, detail="ToDo Not Found")
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
from schema.request import CreateToDoRequest
Base = declarative_base()
# ToDo 클래스 모델링 한 것
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)
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(
# id값은 DB에 의해 자동으로 결정 되서 별도로 지정안해줌
contents=request.contents,
is_done=request.is_done,
)
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List
class ToDoSchema(BaseModel):
id: int
contents: str
is_done: bool
class Config:
from_attributes = True
class ToDoListSchema(BaseModel):
todos: List[ToDoSchema]
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용
from pydantic import BaseModel
class CreateToDoRequest(BaseModel):
contents: str
is_done: bool
