→ FastAPI에서 요청 (request)가 들어왔을 때 session이 생성되어 yield문으로 반환이 되어 사용이 된 후 우리가 응답 (response)을 한 후 이 session을 삭제(close)하는 식으로 FastAPI가 session을 관리함
2. repository pattern 사용 (repository.py 생성 및 쓰기)
목적: repository.py 안에 작성한 함수를 통해 데이터베이스를 조회하는 부분을 생성한 후 이 함수를 main.py에서 사용하기
main.py 안 API에 직접 작성하지 않고 데이터 조회하는 부분을 repository.py라는 새 파일에 작성할 것
이렇게 repository 파일로 코드를 분리하는 것을 repository pattern 이라고 함
0) /projects/todos/src/database/repository.py 파일 생성
# cd /c/Users/관리자/Desktop/projects/todos/src/database
# touch repository.py
1) 데이터 조회하는 코드 작성
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select
from sqlalchemy.orm import Session
from database.orm import ToDo
from typing import List
def get_todos(session: Session) -> List[ToDo]:
return list(session.scalars(select(ToDo)))
from sqlalchemy import select
select 참조
from sqlalchemy.orm import Session
sqlalchemy.orm에서 Session 클래스 참조
from database.orm import ToDo
from database.orm에서 ToDo 클래스 참조
from typing import List
typing에 있는 List 참조; 타입 힌트를 사용하여 명시적으로 타입 정보를 표시하기 위함
def get_todos(session: Session -> List[ToDo]:
get_todos라는 함수 생성
session을 인자로 받음
ToDo를 list에 담아서 반환
return list(session.scalars(select(ToDo)))
전체 ToDo를 조회해서 return을 하게 됨
이 get_todos 함수를 main.py에서 사용할 것
3. 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 # 추가
from database.orm import ToDo # 추가
app = FastAPI()
# 첫 화면 API
@app.get("/")
def health_check_handler():
return {"ping": "pong"}
# POST 생성을 위해 사용자로부터 전달받을 request 클래스 생성
class CreateToDoRequest(BaseModel):
id: int
contents: str
is_done: bool
# 데이터베이스 역할하는 딕셔너리 생성
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),
):
todos: List[ToDo] = get_todos(session=session)
if order and order == "DESC":
return todos[::-1]
return todos
# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(todo_id: int):
todo = todo_data.get(todo_id)
if todo:
return todo
raise HTTPException(status_code=404, detail="ToD Not Found")
# 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]
# 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")
if order and order == "DESC": <\n> return todos[::-1]
원래는 ORM을 사용하여 정렬해주는 것이 좋지만 지금은 생략
여기서는 파이썬에서 코드로 정렬을 해서 정렬함
order 값을 받았고, order 값이 "DESC"이면 todos 목록을 내림차순으로 정렬한 것을 반환
return todos
위의 조건 미충족 경우 todos 목록을 그대로 반환
4. 잘 작동하는지 확인
0) docker 실행 → 데이터베이스 (MySQL) 연결
# 모든 컨테이너 확인
(todos)$ docker ps -a
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 2 weeks ago Exited (0) 5 days ago todos
2597a124c27d docker/welcome-to-docker:latest "/docker-entrypoint.…" 2 weeks ago Exited (0) 2 weeks ago welcome-to-docker
# 컨테이너 시작
(todos)$ docker start 컨테이너명
(todos)$ docker start todos # 나
todos
# 컨테이너 상태 확인
(todos)$ docker ps -a
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 2 weeks ago Up 2 minutes 0.0.0.0:3306->3306/tcp, 33060/tcp todos
2597a124c27d docker/welcome-to-docker:latest "/docker-entrypoint.…" 2 weeks ago Exited (0) 2 weeks ago welcome-to-docker
(todos)$ cd /c/Users/관리자/Desktop/projects/todos/src
3) uvicorn 실행 (uvicorn main:app --reload)
(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 [10676] using StatReload
INFO: Started server process [7916]
INFO: Waiting for application startup.
INFO: Application startup complete.
4) Swagger 문서 확인
브라우저에 http://127.0.0.1:8000/docs 입력
GET /todos Get Todos Handler (전체 todos 조회) 클릭
Try it out 클릭 > Execute 클릭 시
데이터베이스에서 데이터를 가져와 보여준다는 것 확인 (파이썬 딕셔너리에 작성한 content는 한글, 데이터베이스에 작성한 content는 영문 → 영문으로 작성된 content를 보여주기에 데이터베이스에서 갖고온 것을 알 수 있음)
5. 에러
잘 작동 안 될 시 확인할 것들:
docker container run하고 있는지
가상환경 활성화 했는지
경로 올바른 곳에서 uvicorn 실행했는지 (src에서 실행해야 됨)
이때까지의 코드들:
# /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 # 추가
from database.orm import ToDo # 추가
app = FastAPI()
# 첫 화면 API
@app.get("/")
def health_check_handler():
return {"ping": "pong"}
# POST 생성을 위해 사용자로부터 전달받을 request 클래스 생성
class CreateToDoRequest(BaseModel):
id: int
contents: str
is_done: bool
# 데이터베이스 역할하는 딕셔너리 생성
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),
):
todos: List[ToDo] = get_todos(session=session)
if order and order == "DESC":
return todos[::-1]
return todos
# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(todo_id: int):
todo = todo_data.get(todo_id)
if todo:
return todo
raise HTTPException(status_code=404, detail="ToD Not Found")
# 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]
# 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
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})"
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select
from sqlalchemy.orm import Session
from database.orm import ToDo
from typing import List
def get_todos(session: Session) -> List[ToDo]:
return list(session.scalars(select(ToDo)))
