FastAPI를 사용한 파이썬 웹 개발(CHAPTER 6)

앞선 내용에서는 여전히 이벤트 데이터를 저장하기 위해 내장 데이터베이스를 사용하고 있다. 따라서 이제부터는 외부 데이터베이스를 사용하도록 애플리케이션을 변경해보려고 한다.

 

데이터베이스란? 

데이터를 저장하는 창고로, 컬럼(column)(또는 필드(field))과 로우(row)로 구성된 테이블을 갖는다. 

 

내장 데이터베이스의 경우 애플리케이션을 재시작하면 데이터가 모두 지워지기 때문에 데이터를 영구적으로 저장할 수 있는 데이터베이스가 필요하다.

FastAPI를 사용한 파이썬 웹 개발(CHAPTER 6)

CHAPTER 6에서 다루는 내용은 다음과 같다.

• SQLModel 설정
• SQLModel을 사용한 SQL 데이터베이스의 CRUD 처리
• 몽고DB(MongoDB) 설정
• beanie를 사용한 몽고DB의 CRUD 처리

학습을 마치면 FastAPI 애플리케이션을 데이터베이스와 연결할 수 있다. SQLModel을 사용한 SQL 데이터베이스 연결 방법과 beanie를 사용한 몽고DB 데이터베이스 연결 방법을 모두 설명하지만 이후에는 몽고DB를 메인 데이터베이스로 사용한다.

 

<실습 환경>

CHAPTER 6~8의 실습을 진행하려면 몽고DB를 설치해야 한다. 설치 방법은 몽고DB 공식 사이트에서 운영체제에 맞는 문서를 참고하면 된다.

https://www.mongodb.com/docs/manual/installation/ 

 

 

https://smartpoint02.tistory.com/entry/%EB%AA%BD%EA%B3%A0DB-%EC%84%A4%EC%B9%98%ED%95%98%EA%B8%B0

*위의 사이트는 필자가 몽고DB 설치하는 방법을 적은것이다.*

 

https://testdriven.io/authors/adeshina/

*위의 사이트는 저자가 작성한 몽고DB, JWT 인증, 리액트를 활용한 FastAPI 애플리케이션 구축 예제이다.

 

 

6.1 SQLModel 설정

SQL 데이터베이스와 이벤트 플래너 애플리케이션을 연동하려면 먼저 SQLModel 라이브러리를 설치해야 한다.

 

SQL이란?

Structured Query Language의 약자로, 구조적 질의 언어를 뜻한다.

*관계형 데이터베이스 (RDB)에 질의를 하기 위한 언어*

 

SQLModel 라이브러리는 FastAPI 개발자가 만들었으며 pydantic과 SQLAlchemy를 기반으로 한다. pydantic을 통해 모델을 쉽게 정의할 수 있다는 것은 이미 앞에서 알아보았다.

 

SQL과 NoSQL 데이터베이스를 모두 구현해야 하므로 새로운 GitHub 브랜치를 만들 것이다. 먼저 터미널(또는 명령 프롬프트)에서 이벤트 플래너 애플리케이션 폴더로 이동한 다음 Git 저장소를 초기화하고 기존 파일을 커밋한다.

 

$ git init
$ git add database models routes main.py
$ git commit -m "데이터베이스가 없는 기본 애플리케이션 커밋"

 

1

2

3

 

그런 다음 새로운 브랜치를 만든다.

$ git checkout -b planner-sql

필자는 이미 planner-sql 브랜치를&nbsp; 새로 만들었으므로 이미 존재한다고 나온다.

 

SQLModel을 설치할 준비가 끝났다. 터미널에서 가상 환경을 활성화한 후 SQLModel 라이브러리를 설치하자.

필자는 이미 sqlmodel을 설치했으므로 이미 존재한다고 나왔다.

 

 

테이블

테이블이란? 데이터베이스에 저장된 데이터를 가지고 있는 객체이다.

ex) 이벤트 데이터는 이벤트 테이블에 저장된다.

테이블은 컬럼과 로우로 구성되며 이 구조에 맞게 데이터가 저장된다.

 

SQLModel을 사용해서 테이블을 생성하려면 테이블 모델 클래스를 먼저 정의해야 한다. pydantic 모델처럼 테이블을 정의하지만 이번에는 SQLModel의 서브 클래스로 정의한다. 클래스 정의는 table이라는 설정 변수를 갖는다. 이 변수를 통해 해당 클래스가 SQLModel 테이블이라는 것을 인식한다.

 

모델 클래스 안에 정의된 변수는 따로 지정하지 않으면 기본 필드로 설정된다. 만약 필드의 특성을 지정하고 싶다면 Field() 함수를 사용하면 된다. 이벤트 테이블이 어떻게 정의되는지 살펴보자.

 

class Event(SQLModel, table=True):

      id: Optional[int] = Field(default=None, primary_key=True)

      title: str

      image: str

      description: str

      location: str

      tags: List[str]

 

Event 테이블 클래스에서는 id만 필드로 정의되고 나머지는 컬럼으로 정의된다. 필드는 SQLModel 라이브러리가 제공하는 Field 객체를 사용해서 정의된다. id 필드는 테이블의 기본키(primary key)로도 사용된다.

기본키란? 데이터베이스에 저장된 각 레코드를 구분하는 고유한 식별자다.

 

테이블이 무엇인지, 테이블을 어떻게 생성하는지 살펴보았으니 이번에는 로우가 무엇인지 알아보자.

 

로우

데이터베이스 테이블로 전달된 데이터는 특정 컬럼 아래에 있는 로우에 저장된다.

로우에 데이터를 추가하고 저장하려면 테이블의 인스턴스를 만든 후 인스턴스의 변수에 원하는 데이터를 할당하면 된다. 

ex) 하나의 이벤트 데이터를 이벤트 테이블에 추가하려면 다음과 같이 이벤트 모델의 인스턴스를 만들어야 한다.

 

new_event = Event(title="Book Launch",

                               image="src/fastapi.png",

                               description="The book launch event will

                               be held at Packt HQ, Packt city",

                               location="Google Meet",

                               tags=["packt", "book"])

 

그런 다음 세션(Session) 클래스를 사용해서 데이터베이스 트랜잭션(transaction)을 만든다.

 

트랜잭션(Transaction) 이란?  데이터베이스의 상태를 변경시키기 위해 수행하는 작업 단위이다.

데이터베이스의 상태를 변경시킨다는 것은 SELECT, UPDATE, INSERT, DELETE 와 같은 행동을 말한다.

 

이런 트랜잭션은 상황에 따라 여러 개가 만들어질 수 있다.
그 하나의 트랜잭션은 Commit (저장) 되거나 Rollback (철회)될 수 있다.

 

with Session(engine) as session:

      session.add(new_event)

      session.commit()

 

아직 이런 코드가 생소할 것이다. 세션 클래스가 무엇인지, 어떻게 작동하는지부터 알아보자.

 

세션

세션 객체는 코드와 데이터베이스 사이에서 이루어지는 처리를 관리하며 주로 특정 처리를 데이터베이스에 적용하기 위해 사용된다. Session 클래스는 SQL 엔진의 인스턴스를 인수로 사용한다.

 

앞서 테이블과 로우를 어떻게 생성하는지 살펴봤다. 이제 데이터베이스를 어떻게 생성하는지 알아보자.

여기서 사용하는 Session 클래스의 메서드는 다음과 같다.

• add() : 처리 대기중인 데이터베이스 객체를 메모리에 추가한다. 앞서 살펴본 코드에서 new_event 객체는 세션 메모리에 추가되고 commit() 메서드에 의해 데이터베이스에 등록(커밋)될 때까지 대기했다.
• commit(): 현재 세션에 있는 트랜잭션을 모두 정리한다.
• get(): 데이터베이스에서 단일 로우를 추출한다. 모델과 문서 ID를 인수로 사용한다.

지금까지 테이블, 로우, 컬럼 생성 방법과 Session 클래스를 사용해 데이터를 추가하는 방법을 배웠다. 다음 절에서는 데이터베이스를 만들고 CRUD를 처리하는 방법을 살펴본다.

 

6.2 데이터베이스 생성

SQLModel에서는 SQLAlchemy 엔진을 사용해서 데이터베이스를 연결한다. SQLAlchemy 엔진은 create_engine() 메서드를 사용해서 만들며 SQLModel 라이브러리에서 임포트한다.

 

create_engine() 메서드는 데이터베이스 URL을 인수로 사용한다. 데이터베이스 URL은 sqlite:///database.db 또는 sqlite:///database.sqlite와 같은 형식이다. create_engine은 echo를 선택적 인수로 지정할 수 있다. True로 설정하면 실행된 SQL 명령을 출력한다.

 

create_engine() 메서드만으로는 데이터베이스 파일을 만들 수 없다. SQLModel.metadata.create_all(engine) 메서드를 사용해서 create_engine() 메서드의 인스턴스를 호출해야 한다.

 

database_file = "database.db"

engine = create_engine(database_file, echo=True)

SQLModel.metadata.create_all(engine)

 

create_all() 메서드는 데이터베이스뿐만 아니라 테이블도 생성한다. 중요한 점은 데이터베이스 연결 파일(connection.py)에서 테이블 파일을 임포트해야 한다는 것이다.

 

이벤트 플래너 애플리케이션에서는 이벤트에 대한 CRUD 처리를 수행한다. database 폴더에 다음과 같이 파일을 생성하자. 이 파일에 데이터베이스 연결을 위한 데이터를 설정할 것이다.

 

(venv) $ touch database/connection.py

 

데이터베이스 연결을 위한 파일을 생성했으니 다음 단계를 따라 애플리케이션을 데이터베이스와 연동하는 함수를 작성하자.

 

1. {models/events.py}에 정의한 이벤트 모델 클래스(Event)를 변경해서 SQLModel의 테이블 클래스를 사용하도록 만든다.

 

from sqlmodel import JSON, SQLModel, Field, Column
from typing import Optional, List

 

class Event(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    title: str
    image: str
    description: str
    tags: List[str] = Field(sa_column=Column(JSON))
    location: str

 

   class Config:
        arbitrary_types_allowed = True
        schema_extra = {
                 "example": {
                       "title": "FastAPI Book Launch",
                       "image": "https://linktomyimage.com/image.png",
                       "description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",
                       "tags":["python","fastapi","book","launch"],
                       "location": "Google Meet"
        }
    }

 

 

이 코드는 기존 모델 클래스를 SQL 테이블 클래스로 변경한다.

 

2. UPDATE 처리의 바디 유형으로 사용할 SQLModel 클래스를 추가한다. { models 폴더의 events.py }

 

class EventUpdate(SQLModel):
    title: Optional[str]
    image: Optional[str]
    description: Optional[str]
    tags: Optional[List[str]]
    location: Optional[str]

    class Config:
        schema_extra = {
            "example": {"title": "FastAPI Book Launch",
            "image": "https://linktomyimage.com/image.png",
            "description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",
            "tags": ["python", "fastapi", "book", "launch"],
            "location": "Google Meet"
          }
        }

 

3. 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)

 

def get_session():

      with Session(engine_url) as session:

            yield session

 

이 코드는 먼저 필요한 의존 라이브러리와 테이블 모델을 임포트한다. 그리고 데이터베이스 파일의 위치(없는 경우 생성된다), 연결 문자열(connection string), 생성된 SQL 데이터베이스의 인스턴스를 변수에 저장한다. conn() 함수는 SQLModel을 사용해서 데이터베이스와 테이블을 생성하고 get_session()을 사용해서 데이터베이스 세션을 애플리케이션 내에서 유지한다.

 

4. main.py를 다음과 같이 변경하여 애플리케이션이 시작될 때 데이터베이스를 생성하도록 한다.

 

from fastapi import FastAPI

from fastapi.responses import RedirectResponse

from typing import List

from database.connection import conn

 

from routes.users import user_router
from routes.events import event_router

 

import uvicorn 

app = FastAPI()

# 라우트 등록
app.include_router(user_router, prefix="/user")
app.include_router(event_router, prefix="/event")

@app.on_event("startup")
def on_startup():

     conn()    
    
@app.get("/")
async def home():
    return RedirectResponse(url="/event/")

if __name__ == '__main__':
    uvicorn.run("main:app", host="127.0.0.1", port=8000 , reload=True)

 

+++

RedirectResponse는 상태 코드 307(리다이렉트)을 반환한다. 여기서는 "/"으로 접속한 경우 "/event"로 리다이렉트하기 위해 사용된다.

+++

 

===

책에서 변경하라고(추가하라고) 한 부분

추가적으로 추가되어야 할 것

===

 

 

애플리케이션을 실행하면 데이터베이스가 생성된다. 시작 시 conn() 함수를 호출해서 데이터베이스를 생성하기 때문이다. 터미널에서 애플리케이션을 실행하면 데이터베이스와 테이블이 생성됐다는 메시지가 출력된다.

 

사진 1 (필자가 임의로 지은 이름)

사진 2 (필자가 임의로 지은 이름 ) [&ne; 사진 1]

 

사진 3 (필자가 임의로 지은 이름)

 

사진 1 과 사진 2 와 사진 3 에서 볼 수 있듯이 SQL 명령이 화면에 출력되는 이유는 데이터베이스 엔진을 만들 때 echo를 True로 설정했기 때문이다.

 

데이터베이스를 생성했으니 데이터베이스를 사용하도록 CRUD 처리 라우트를 변경해보자.

 

이벤트 생성

routes/events.py를 변경해서 이벤트 테이블 클래스와 get_session() 함수를 임포트한다. get_session() 함수를 통해 라우트가 세션 객체에 접근할 수 있다.

 

from fastapi import APIRouter, Depends, HTTPException, Request, status

from database.connection import get_session
from models.events import Event, EventUpdate

 

**

Depends 클래스는 FastAPI 애플리케이션에서 의존성 주입(dependency injection)을 담당한다. 이 클래스는 함수를 인수로 사용하거나 함수 인수를 라우트에 전달할 수 있게 해서 어떤 처리가 실행되든지 필요한 의존성을 확보해준다.

**

 

신규 이벤트 생성을 담당하는 POST 라우트를 다음과 같이 변경한다.

 

@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"
    }

 

이 코드는 데이터베이스 처리에 필요한 세션 객체가 get_session() 함수에 의존하도록 설정한다. 함수 내에서는 데이터(이벤트)를 세션에 추가하고 데이터베이스에 등록(커밋)한 후 세션을 업데이트한다.

 

다음 명령을 사용해 라우트가 제대로 변경됐는지 확인한다

**처리가 실패한 경우에는 예외가 반환된다**

 

이벤트 조회

전체 이벤트를 추출하는 GET 라우트를 변경해서 데이터베이스에서 데이터를 가져오도록 만들어보자. 

다음과 같이 routes/events.py 파일을 변경하면 된다.

 

from sqlmodel import select

 

@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).all()

       return events

 

 

마찬가지로 지정한 ID의 이벤트 정보를 표시하는 라우트도 변경한다.

 

@event_router.get("/{id}", response_model=Event)
async def retrieve_event(id: int, session=Depends(get_session)) -> Event:

        event = session.get(Event, id)

        if event:

            return event

 

        raise HTTPException(

             status_code=status.HTTP_404_NOT_FOUND,

             detail="Event with supplied ID does not exist"

        )

 

 

두 라우트의 응답 모델이 모델 클래스를 사용하도록 설정했다.

 

이제 이벤트 목록을 반환하는 GET 요청을 테스트해보자. 

 

이번에는 ID를 지정해서 이벤트를 추출해보자.

 

 

 

****

현재 이벤트가 하나뿐이기 떄문에 두 라우트의 결과가 같다.

 

****

 

///////

(다음날)

서버 돌린다!!!

DELETE와 GET은 실습한 내용임

//////

 

이벤트 변경

routes/events.py 파일에 UPDATE 라우트를 다음과 같이 추가한다.

 

@event_router.put("/edit/{id}", response_model=Event)
async def update_event(id: int, new_data: EventUpdate, session=Depends(get_session)) -> Event:

 

라우트 함수에 특정 이벤트를 추출해서 변경하는 코드를 작성한다.

    event = session.get(Event, id)
    if event:
        event_data = new_data.dict(exclude_unset=True)
        for key, value in event_data.items():
            setattr(event, key, value)
        session.add(event)
        session.commit()
        session.refresh(event)

        return event
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Event with supplied ID does not exist"
    )

 

이 코드는 이벤트를 변경하기 전에 해당 이벤트가 존재하는지를 먼저 확인한다. 이벤트를 변경한 후에는 변경한 데이터를 반환한다. 다음 명령을 사용해 기존 이벤트의 타이틀(title)을 변경해보자.

 

 

이벤트 삭제

events.py 파일에서 기존 DELETE 라우트를 다음과 같이 변경한다. 

 

@event_router.delete("/delete/{id}")
async def delete_event(id: int, session=Depends(get_session)) -> dict:
    event = session.get(Event, id)
    if event:
        session.delete(event)
        session.commit()
        return {
            "message": "Event deleted successfully"
            }

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Event with supplied ID does not exist"
        )

 

이 코드는 해당 ID의 이벤트가 존재하는지 확인한 후 존재하면 데이터베이스에서 삭제한다.

 

처리가 끝나면 성공 메시지 또는 이벤트가 존재하지 않는다는 메시지를 반환한다.

데이터베이스에서 이벤트 하나를 삭제해보자.

 

 

전체 이벤트를 추출하는 명령을 실행해보자.

 

명령을 실행하면 다음과 같이 빈 배열이 반환된다. 기존 이벤트를 삭제했으므로 등록된 이벤트가 없기 때문이다.

[]

 

 

====

 

갑자기 또 서버가 오류가 났다.ㅠㅠ

아무리 POST 명령을 해도 서버 에러가 났다.

근데 DELETE와 GET(DELETE 후 전체 이벤트 추출) 명령을 할 때에는 정상적으로 (요청이) 처리되었다.

문제를 (어찌저찌(?))해결했더니 드디어!! 정상적으로 (요청이) 처리되었다~~~~

 

====

 

지금까지 SQLModel을 사용해서 애플리케이션과 데이터베이스를 연결하고 CRUD 처리를 구현했다. 작성한 코드를 커밋하자.

 

1

1

그런 다음 브랜치를 마스터 브랜치로 변경한다.

1

 

 

 

********

 

2

********

 

 

6.3 몽고DB 설정

FastAPI와 몽고DB를 연결해주는 몇 가지 도구가 존재하지만 여기서는 beanie를 사용한다. 

beanie는 비동기 객체 문서 매퍼(Object Document Mapper(ODM))로, 데이터베이스 처리를 담당한다.

 

다음 명령을 사용해서 beanie를 설치한다.

(venv)$ pip install beanie==1.13.1

 

*예제 코드를 실행하려면 beanie를 1.13.x 버전으로 지정해야 한다. 1.14.x부터는 실행되지 않는다.*

 

데이터베이스 연동에 앞서 beanie의 기능과 데이터베이스 테이블 생성 방법을 알아보자.

 

문서

SQL에서는 데이터가 컬럼과 로우로 구성된 테이블에 저장된다. NoSQL 데이터베이스에서는 데이터 저장을 위해 문서를 사용한다. 문서는 데이터베이스 컬렉션에 데이터가 저장되는 형식으로, pydantic 모델과 동일한 방식으로 정의된다. 유일한 차이점은 beanie가 제공하는 Document 클래스를 사용한다는 점이다.

 

예를 들면 다음과 같이 문서를 정의할 수 있다.

 

from beanie import Document

 

class Event(Document):

      name: str

      location: str

 

      class Settings:

            name = "events"

 

여기서 Settings 서브 클래스는 몽고DB 데이터베이스 내에 설정한 이름으로 컬렉션을 생성한다.

 

문서 생성 방법을 알았으니 CRUD 처리를 위한 메서드를 살펴보자.

 

• insert() , create() : 문서 인스턴스에 의해 호출되며 데이터베이스 내에 새로운 레코드를 생성한다. 단일 데이터는 insert_one() 메서드를 사용해 추가하고 여러 개의 데이터는 insert_many() 메서드를 사용해 추가한다.

       event1 = Event(name="Packt office launch", location="Hybrid")

       event2 = Event(name="Hanbit office launch", location="Hybrid")

       await event1.create()

       await event2.create()

       await Event.insert_many([event1, event2])

 

• find() , get() : find() 메서드 = 인수로 지정한 문서를 문서 목록에서 찾는다.  get() 메서드= 지정한 ID와 일치하는 단일 문서를 반환한다. find_one() 메서드 =  다음과 같이 지정한 조건과 일치하는 단일 문서를 반환한다.

       event = await Event.get("74478287284ff")

       # 일치하는 아이템의 리스트를 반환한다.

       event = await Event.find(Event.location == "Hybrid").to_list()

       # 단일 이벤트를 반환한다.

       event = await.find_one(Event.location == "Hybrid")

 

++   "74478287284ff"는 몽고DB의 고유한 문서 ID를 의미한다.  ++

 

• save() , update() , upsert() : save() 메서드 = 데이터를 신규 문서로 저장할 때 사용된다. update() 메서드 = 기존 문서를 변경할 때 사용된다. upsert() 메서드 = 조건에 부합하는 문서가 없으면 신규로 추가할 때 사용된다.(즉, 변경하려는 문서가 없으면 새롭게 추가한다. update와 insert를 합친 용어다) .  여기서는 update() 메서드를 사용하며 다음과 같이 변경용 쿼리를 지정한다.

       event = await Event.get("74478287284ff")

       update_query = {"$set": {"location": "virtual"}}

       await event.update(update_query)

 

       이 코드는 변경하고자 하는 쿼리를 추출한 후 해당 문서의 location 필드를 virtual (온라인)로 변경한다.

 

  

• delete() : 데이터베이스에서 문서를 삭제한다. 사용 방법은 다음과 같다.

       event = await Event.get("74478287284ff")

       await event.delete()

 

지금까지 beanie 라이브러리가 제공하는 메서드를 살펴봤다. 이제 이벤트 플래너 애플리케이션에 몽고DB를 설정하고 문서를 정의해보자.

 

데이터베이스 초기화

데이터베이스 초기화 과정을 단계별로 살펴보자.

 

1. database 폴더에 connection.py라는 파일을 만든다.

(venv)$ touch connection.py

 

 

pydantic의 BaseSettings 부모 클래스를 사용해서 설정 변수를 읽을 수 있다. 웹 API를 개발할 때는 설정 변수를 하나의 환경 파일에 저장하는 것이 좋다.

 

2. connection.py에 다음 코드를 추가한다.

 

from beanie import init_beanie

from motor.motor_asyncio import AsyncIOMotorClient

from typing import Optional

from pydantic import BaseSettings

 

class Settings(BaseSettings):

       DATABASE_URL: Optional[str] = None

 

       async def initialize_database(self):

            client = AsyncIOMotorClient(self.DATABASE_URL)

            await init_beanie(database=client.get_default_database(), 

                                         document_models=[])

 

       class Config:

              env_file = ".env"

 

이 코드는 먼저 데이터베이스를 초기화하기 위해 필요한 라이브러리를 임포트한다. 그런 다음 Settings 클래스를 정의해서 데이터베이스 URL(DATABASE_URL)을 설정한다. 데이터베이스 URL은 Config 서브 클래스에 정의된 환경 파일(env_file)에서 읽어온다. 마지막으로 initialize_database() 메서드를 정의해서 데이터베이스를 초기화한다.

 

init_beanie() 메서드는 데이터베이스 클라이언트를 설정한다. SQLModel에서 생성한 몽고 엔진 버전과 문서 모델을 인수로 설정한다.

 

3. models 폴더의 모델 파일을 변경하여 몽고DB 문서를 사용할 수 있도록 만들자. 다음과 같이 models/events.py 파일을 변경한다.

 

from beanie import Document

from typing import Optional, List

 

class Event(Document):

        title: str

        image: str

        description: str

        tags: List[str]

        location: str

 

        class Config:

               schema_extra = {

                       "example": {

                               "title": "FastAPI Book Launch",

                               "image": "https://linktomyimage.com/image.png",

                               "description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",

                                "tags": ["python", "fastapi", "book", "launch"],

                                "location": "Google Meet"

                         }

                 }

           class Settings:

                  name =  "events"

 

 

 

4. UPDATE 처리를 위한 pydantic 모델을 동일한 파일에 추가한다.

 

class EventUpdate(BaseModel):

       title: Optional[str]

       image: Optional[str]

       description: Optional[str]

       tags: Optional[List[str]]

       location: Optional[str]

 

       class Config:

              schema_extra = {

                       "example": {

                                "title": "FastAPI Book Launch",

                                "image": "https://linktomyimage.com/image.png",

                                "description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",

                                 "tags": ["python", "fastapi", "book", "launch"],

                                 "location": "Google Meet"

                        }

              }

 

 

5. models/users.py 파일을 다음과 같이 변경한다.

 

from typing import Optional, List

from beanie import Document, Link

from pydantic import BaseModel, EmailStr

from models.events import Event

 

class User(Document):

       email: EmailStr

       password: str

       events: Optional[List[Event]]

 

       class Settiongs:

              name = "users"

 

        class Config:

              schema_extra = {

                       "example": {

                                "email": "fastapi@packt.com",

                                "username": "strong!!!",

                                "events": [],

                         }

               }

 

class UserSignIn(BaseModel):

       email: EmailStr

       password: str

 

 

 

6. (database의) connection.py의 document_models 필드를 다음과 같이 변경한다.

 

from models.users import User

from models.events import Event  (이미 파일에 코드 존재)

 

async def initialize_database(self):

         client = AsyncIOMotorClient(self.DATABASE_URL)

         await init_beanie(database=client.get_default_database(),

                document_models=[Event, User])

 

class Config:

       env_file = ".env"

 

7. 환경 파일(.env)을 생성한 다음 데이터베이스 URL을 추가하면 데이터베이스 초기화 과정이 끝난다.

 

(venv)$ touch .env

(venv_$ echo DATABASE_URL=mongodb://localhost:27017/planner >> .env

 

 

 

데이터베이스 초기화 작업이 모두 끝났다. 다음으로 CRUD 처리를 구현해보자.

 

6.4 CRUD 처리

 

connection.py 파일에 다음과 같이 새로운 Database 클래스를 추가한다. 이 클래스는 초기화 시 모델을 인수로 받는다.

 

from pydantic import BaseSettings, BaseModel

from typing import Any, List, Optional

from beanie import init_beanie, PydanticObjectId

 

class Database:

       def  __init__(self, model):

              self.model = model

 

데이터베이스 초기화 중에 사용되는 모델은 Event 또는 User 문서의 모델이다.

 

생성 처리

 

Database 클래스 안에 다음과 같이 save() 메서드를 추가한다. 이 메서드는 레코드 하나를 데이터베이스 컬렉션에 추가한다.

 

       async def save(self, document) -> None:

               await document.create()

               return

 

이 코드는 문서를 인수로 받는 save() 메서드를 정의한다. 정확히 말하면 문서의 인스턴스를 받아서 데이터베이스 인스턴스에 전달한다.

 

조회처리

 

데이터베이스 컬렉션에서 단일 레코드를 불러오거나 전체 레코드를 불러오는 메서드를 작성해보자.

 

       async def get(self, id: PydanticObjectId) -> Any:

               doc = await self.model.get(id)

               if doc:

                     return doc

               return False

 

       async def get_all(self) -> List[Any]:

               docs = await self.model.find_all().to_list()

               return docs

 

첫 번째 메서드 get()은 ID를 인수로 받아 컬렉션에서 일치하는 레코드를 불러온다. 반면 두 번째 메서드 get_all()은 인수가 없으며 컬렉션에 있는 모든 레코드를 불러온다.

 

변경 처리 

 

기존 레코드를 변경하는 메서드를 작성해보자.

 

      async def update(self, id: PydanticObjectId, body: BaseModel) -> Any:

              doc_id = id

              des_body = body.dict()

              des_body = {k:v for k,v in des_body.items() if v is

              not None}

              update_query = {"$set": {

                       field: value for field, value in

                       des_body.items()

               }}

 

              doc = await self.get(doc_id)

              if not doc:

                   return False

              await doc.update(update_query)
              return doc

 

 

update() 메서드는 하나의 ID와 pydantic 스키마(모델)를 인수로 받는다. 스키마에는 클라이언트가 보낸 PUT 요청에 의해 변경된 필드가 저장된다. 변경된 요청 바디는 딕셔너리에 저장된 다음 None값을 제외하도록 필터링된다. 이 작업이 완료되면 변경 쿼리에 저장되고 beanie의 update() 메서드를 통해 실행된다.

 

삭제 처리

 

데이터베이스에서 레코드를 삭제하는 메서드를 작성해보자.

 

      async def delete(self, id: PydanticObjectId) -> bool:

              doc = await self.get(id)

              if not doc:

                    return False

              await doc.delete()

              return True

 

이 메서드는 먼저 해당 레코드가 있는지 확인하고 있으면 삭제한다.

 

CRUD 처리용 메서드가 추가된 데이터베이스 파일을 완성했다. 이제 라우트를 변경해보자.

 

 

routes/events.py

 

import문과 database 인스턴스를 다음과 같이 변경한다.

 

from beanie import PydanticObjectId

from fastapi import APIRouter, HTTPException, status

from database.connection import Database

 

from models.events import Event, EventUpdate

from typing import List

event_database = Database(Event)

 

모든 라우트를 변경해야 한다. GET 라우트부터 시작하자. GET 라우트는 앞서 데이터베이스 파일에 정의한 메서드를 호출한다.

 

@event_router.get("/", response_model=List[Event])

async def retrieve_all_events() -> List[Event]:

        events = await event_database.get_all()

        return events

 

@event_router.get("/{id}", response_model=Event)

async def retrieve_event(id: PydanticObjectId) -> Event:

        event = await event_database.get(id)

        if not event:

              raise HTTPException(

                     status_code=status.HTTP_404_NOT_FOUND,

                     detail="Event with supplied ID does not exist"

               )

         return event

 

 

POST 라우트를 다음과 같이 변경한다.

 

@event_router.post("/new")

async def create_event(body: Event) -> dict:

         await event_database.save(body)

         return {

                "message": "Event created successfully."

         }

 

 

UPDATE 라우트를 다음과 같이 추가한다.

 

@event_router.put("/{id}", response_model=Event)

async def update_event(id: PydanticObjectId, body: EventUpdate)-> Event:

         updated_event = await event_database.update(id, body)

         if not updated_event:

               raise HTTPException(

                      status_code=status.HTTP_404_NOT_FOUND,

                      detail="Event with supplied ID does not exist"

               )

         return updated_event

 

마지막으로 DELETE 라우트를 다음과 같이 변경한다.

 

@event_router.delete("/{id}")

async def delete_event(id: PydanticObjectId) -> dict:

        event = await event_database.delete(id)

        if not event:

              raise HTTPException(

                     status_code=status.HTTP_404_NOT_FOUND,

                     detail="Event with supplied ID does not exist"

              )

         return {

               "message": "Event deleted successfully."

         }

 

이벤트 라우트용 CRUD 처리를 모두 구현했다. 이어서 사용자 등록 및 로그인 라우트를 구현해보자.

 

routes/users.py

 

import문과 데이터베이스 인스턴스 생성 코드를 다음과 같이 변경한다.

 

from fastapi import APIRouter, HTTPException, status

from database.connection import Database

 

from models.users import User, UserSignIn

 

user_router = APIRouter(

        tags=["User"],

)

 

user_database = Database(User)

 

사용자 등록용 POST 라우트를 다음과 같이 변경한다.

 

@user_router.post("/signup")

async def sign_new_user(user: User) -> dict:

        user_exist = await User.find_one(User.email == user.email)

        if user_exist:

              raise HTTPException(

                    status_code=status.HTTP_409_CONFLICT,

                    detail="User with email provided exists already"

              )

       await user_database.save(user)

       return {

             "message": "User created successfully."

       }

 

해당 이메일의 사용자가 존재하는지 확인하고 없으면 데이터베이스에 등록한다.

 

이번에는 로그인 라우트를 다음과 같이 변경하자.

 

@user_router.post("/signin")

async def sign_user_in(user: UserSignIn) -> dict:

        user_exist = await User.find_one(User.email == user.email)

        if not user_exist:

             raise HTTPException(

                    status_code=status.HTTP_404_NOT_FOUND,

                    detail="User with email does not exist"

             )

         if user_exist.password == user.password:

              return {

                     "message": "User signed in successfully."

              }

         raise HTTPException(

               status_code=status.HTTP_401_UNAUTHORIZED,

               detail="Invalid details passed"

         )

 

이 라우트는 먼저 해당 사용자가 존재하는지 확인한다. 여기서는 아주 간단한 사용자 인증 방법을 사용하는데 실제 프로덕션에는 권장하지 않는 방법이다. 인증은 <CHAPTER 7 보안>에서 자세히 다룬다.

 

마지막으로 main.py를 수정해서 애플리케이션 실행 시 몽고DB를 초기화하도록 만든다. 강조된 코드를 추가하면 된다.

 

from database.connection import Settings

 

app = FastAPI()

settings = Settings()

# 라우트 등록

app.include_router(user_router,   prefix="/user")

app.include_router(event_router, prefix="/event")

 

@app.on_event("startup")

async def init_db():

        await settings.initialize_database()

 

그런 다음 dotenv 라이브러리를 추가로 설치한다.

 

(venv) $ pip install pydantic[dotenv]

 

 

이것으로 라우트 구현이 끝났다. 이제 몽고DB 인스턴스와 이벤트 플래너 애플리케이션을 실행해보자. 몽고DB 데이터베이스가 상주할 폴더를 생성한 후 몽고DB 인스턴스를 시작하면 된다.

*몽고DB가 설치되어 있어야 한다. 이 글의 앞부분 내용 참고하세요!*

 

(venv)$ mkdir store

(venv)$ mongod --dbpath store

또 다른 터미널 창을 열어서 애플리케이션을 실행한다.

 

 

다음 단계에 따라 이벤트를 테스트해보자.