[book] fastapi를 사용한 파이썬 웹 개발 - 학습 노트

 - 해당 책은 번역서로, 저장소가 2개가 있다. 하나는 해외 저자가 만든 저장소이고 하나는 한빛에서 만든 저장소이다.

 - 저자의 저장소는 오류가 있다. 한빛 저장소를 사용한다.

 - 저장소 : https://github.com/hanbit/web-with-fastapi.git

GitHub - hanbit/web-with-fastapi: 『FastAPI를 사용한 파이썬 웹 개발』(한빛미디어, 2023) 예제 코드 저장소

1. 중첩 모델

   - 중첩 모델을 다음과 같이 선언할 수 있다.

class Item(BaseModel):
    item: str


class Todo(BaseModel):
    id: int
    item: Item

 

   - 호출은 다음과 같은 json 형식을 사용해야 한다.

{
	"id":1,
    "item" :{
 		"item":"first commit"
	}
}

 

 2. BaseModel의 schema_extra 변경 사항

   - 책에서 사용되는 schema_extra의 선언 방법은 다음과 같다.

    class Config:
        schema_extra = {
            "examples": [{
                "id": 1,
                "item": "Example Schema!"
            }]
        }

 

   - Python 3.10+ Pydantic v2 에서는 다음과 같이 사용해야 한다.

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "id": 1,
                    "item": "Example Schema!"
                }
            ]
        }
    }

 

   - 적용 결과 확인

      - http://127.0.0.1:8000/redoc 을 접속하여 add Todo를 누르면 아래와 같은 결과를 창에서 확인할 수 있다.

 

   - reference : https://fastapi.tiangolo.com/tutorial/schema-extra-example/

Declare Request Example Data - FastAPI

 3. for을 이용해 list 안에 있는 dict type의 데이터중 일부를 확인하고 해당 dict를 삭제하는 방법

   - 책에서 사용한 for이 마음에 들지 않아 다른 방법과 함께 기록한다.

      - 책에서 사용한 방법

    for index in range(len(todo_list)):
        todo = todo_list[index]
        if todo.id == todo_id:
            todo_list.pop(index)
            return {"message": "Todo deleted successfully."}

 

      - 내가 사용한 방법

    for index, item in enumerate(todo_list):
        if item.id == todo_id:
            del todo_list[index]
            return {"message": "Todo deleted successfully."}

 

4. 문서화 페이지에서 api hint 적용 확인 방법

   - docs : http://127.0.0.1:8000/docs

 

   -  redoc : http://127.0.0.1:8000/redoc

 

5. 옵셔널 필드와 form 설정

   - 현재 web page에서 id를 정의하지 않고 기존 id에서 1씩 더해 자동으로 증가하도록 만들고 있다.

   - 기존 코드

class Todo(BaseModel):
    id: Optional[int]
    item: str

    @classmethod
    def as_form(cls, item: str = Form(...)):
        return cls(item=item)

 

   - 기존 코드를 실행하면 다음과 같은 에러가 발생했다.

pydantic_core._pydantic_core.ValidationError: 1 validation error for Todo
  Field required [type=missing, input_value={'item': 'asdf'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.5/v/missing

 

   - 에러 발생 원인은 optional 필드에 대해 form에서 처리를 해주지 않았기 때문이다

   - 해결된 코드

class Todo(BaseModel):
    id: Optional[int]
    item: str

    @classmethod
    def as_form(cls, item: str = Form(...)):
        return cls(id=None, item=item)

 

6. databse 연결 : sqlalchemy

   - 책에서의 내용이 매우 부실하여 상세 설명이 어렵다. 대략적인 처리 과정에 대해서만 기록한다.

   - main.py

# 서비스 시작과 동시에 main.py에서 databse 생성 시도
@app.on_event("startup")
def on_startup():
    conn()

 

   - connection.py

      - 데이터 베이스 생성 및 연결

from sqlmodel import SQLModel, Session, create_engine
from models.events import Event

database_file = "planner.db"
database_connection_string = f"sqlite:///{database_file}"
connect_args = {"check_same_thread": False}
engine_url = create_engine(
    database_connection_string, echo=True, connect_args=connect_args
)

# 서비스의 시작과 동시에 데이터베이스 생성
def conn():
    SQLModel.metadata.create_all(engine_url)
    
# db 연결이 필요한 모든 라우터의 실행시, 
# 라우터로 정의된 함수의 session=Depends(get_session) 인수를 사용해 연결
def get_session():
    with Session(engine_url) as session:
        yield session

 

   - events.py : 라우터 연결

      - 데이터 저장 및 가져오기

      - 책에서는 select(Event)한 값을 session.exec(statement).all() 로 호출하라 명시되어 있다. 하지만 에러가 발생한다 이것은 시리얼라이저와 관련된 사항으로 반환된 쿼리의 format이 models.py에 정의된 Evant와 적합하지 않아서 발생된다.

         - 요구되는 필드를 직접 하나씩 정의하거나, scalars()를 사용하면 해결이 된다. 

      - 기본적으로 Column를 사용한 필드 정의와 basemodel과 pydantic의 사용방법,  class config의 사용 방법을 더 학습해야 할 필요가 있다.

      - orm을 사용하여 보다 복잡한 sql을 만들어 사용할 수 있다.

@event_router.post("/new")
async def create_event(new_event: Event, session=Depends(get_session)) -> dict:
    session.add(new_event)
    session.commit()
    session.refresh(new_event)

    return {"message": "Event created successfully"}


@event_router.get("/", response_model=List[Event])
async def retrieve_all_events(session=Depends(get_session)) -> List[Event]:
    statement = select(Event)
    events = session.exec(statement).scalars().all()
    return events

 

   - json으로 업데이트 요청을 할 경우 다음과 같은 값을 보내야 한다.

      - title에 대한 값만 업데이트 하려고 하는 경우, 나머지 값에 대해서 key는 항상 존재해야 한다.

{
    "title": "Packts FastAPI book launch 2",
    "image": "",
    "description": "",
    "tags": [],
    "location": ""
}

 

 * 개발 시간 단축을 위해 책을 구매하고 학습을 한 목적이 있는데, 프로젝트 코드 누락과 책 내용상 코드 오류가 난무한 상황과 책의 코드에 대해 주석이나 설명이 거의 전무하다 싶은 상황에 얆은 책 한권 읽는 시간이 검색과 추가 학습 그리고 코드 검증으로 기대했던 기간의 수배가 들어가고 있기에 더 이상 코드로 테스트를 중단하고 나머지 부분은 눈으로 대략적인 흐름과 이해를 하고 마무리 하기로 했다.

 

 - 이후 항목 중 pytest는 차후 참고할 여기는 있어보였다.

 - 리뷰가 5점 만점인 책, 목차가 나름 체계가 있어보여 샀지만, 뭐랄까... 리뷰 체험단의 실체를 이제야 당한 기분이 크다.

 - 오프라인에서 확인하기 어려운 기술서적은 대체 어떻게 확인을 해야할지... 많이 실망한 채 글을 마무리 한다.

 

 - reference : 

https://soogoonsoogoonpythonists.github.io/sqlalchemy-for-pythonist/

파이썬 개발자를 위한 SQLAlchemy

https://velog.io/@jihwankim94/Python-Sqlalchemy-%EA%B5%AC%EC%A1%B0

[Python] Sqlalchemy 구조 ORM, Core 에 대해

https://sonsazang.tistory.com/31

[FastAPI, MongoDB] No default database name defined or provided 버그 수정