# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select
from sqlalchemy.orm import Session
from typing import List
from database.orm import ToDo
...
# 단일 ToDo 조회하는 함수 생성
def get_todo_by_todo_id(session: Session, todo_id: int) -> ToDo | None:
# 이 type hint는 Python 3.10 버전 문법
# Python 3.10 버전보다 아래 버전은 Optional을 import하여 아래와 같이 작성
# def get_todo_by_todo_id(session: Session, todo_id: int) -> Optional[ToDo]:
return session.scalar(select(ToDo).where(ToDo.id == todo_id))
def get_todo_by_todo_id(session: Session, todo_id: int) -> ToDo | None:
즉, scalar 메소드에 의해 ToDo가 존재하는 경우에는 ToDo를 반환, todo_id에 해당하는 ToDo가 존재하지 않는 경우에는 none을 반환
이제 이 get_todo_by_todo 함수를 main.py에서 이용하도록 할 것
2. 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 # 추가
from database.orm import ToDo
from schema.response import ToDoSchema, ListToDoResponse
...
# 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")
# GET Method 사용하여 단일 조회 API - DB 참조
@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 Found")
from database.repository import get_todos, get_todo_by_todo_id
새로 생성한 get_todo_by_todo_id 참조
def get_todo_handler( ... ) -> ToDoSchema:
type hint를 사용하여 반환 값으로 ToDoSchema를 반환한다는 것을 명시
todo: ToDo | None = get_todo_by_todo_id(session=session, todo_id=todo_id)
type hint를 통해 todo라는 변수에 ToDo 또는 None을 할당 받을 것을 명시
todo는 session과 todo_id를 전달 받아 얻어진 get_todo_by_todo_id를 할당 받을 것임
session은 Depends(get_db)를 통해 session=session에 전달
todo_id는 위 @app.get("/todos/{todo_id}", status_code=200) 안에 {todo_id}에 있는 path를 통해 todo_id=todo_id 에 전달
if todo:
todo가 있는 경우
return ToDoSchema.from_orm(todo)
ToDoSchema가 그대로 반환됨 (전체 조회일 때에는 리스트 형태로 반환 됐었음)
ToDo ORM 객체가 ToDoSchema로 변경이 되어 반환
5. 실습으로 확인 - 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 입력
GET /todos/{todo_id} Get Todo Handler 클릭 > Try it out > Parameters 아래 todo_id 값에 1 입력 > Execute
1번 todo가 정상적으로 출력된 것 확인
존재 하지 않는 값 (4번) 입력 시
Responses 아래 Code와 Detail에 "404"와 "Error: Not Found"가 출력되는 것을 확인
# /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 # 추가
from database.orm import ToDo
from schema.response import ToDoSchema, ListToDoResponse
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),
) -> ListToDoResponse:
todos: List[ToDo] = get_todos(session=session)
if order and order == "DESC":
# return todos[::-1]
return ListToDoResponse(
todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
)
# return todos
return ListToDoResponse(
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):
# todo = todo_data.get(todo_id)
# if todo:
# return todo
# raise HTTPException(status_code=404, detail="ToD Not Found")
@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_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 typing import List
from database.orm import ToDo
def get_todos(session: Session) -> List[ToDo]:
return list(session.scalars(select(ToDo)))
def get_todo_by_todo_id(session: Session, todo_id: int) -> ToDo | None:
return session.scalar(select(ToDo).where(ToDo.id == todo_id))
# /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:
orm_mode = True
class ListToDoResponse(BaseModel):
todos: List[ToDoSchema]
