Ubuntu 24.04에서 MongoDB와 함께 Fast API 설치하는 방법

FastAPI는 API 서비스를 생성하기 위해 Python을 기반으로 한 웹 프레임워크입니다. 현대적이고 빠르며 고성능의 프레임워크로 비동기 작업을 지원합니다.

이 튜토리얼에서는 Ubuntu 24.04에서 MongoDB와 함께 FastAPI를 설치하는 방법을 배웁니다. 또한 FastAPI와 MongoDB 데이터베이스를 사용하여 CRUD 작업을 수행하는 첫 번째 API를 만드는 방법도 배웁니다.

전제 조건

이 가이드를 시작하기 전에 다음 사항을 확인하세요:

• Ubuntu 24.04 시스템
• 관리자 권한이 있는 비루트 사용자

MongoDB 설치

새로운 FastAPI 프로젝트를 생성하기 전에 시스템에 MongoDB 서버를 설치합시다.

먼저, 아래 명령어를 실행하여 패키지 인덱스를 업데이트하고 ‘gnupg’와 ‘curl’을 시스템에 설치합니다.

sudo apt update && sudo apt install gnupg curl -y

아래 명령어를 실행하여 MongoDB 서버의 GPG 키를 추가합니다.

curl -fsSL https://www.mongodb.org/static/pgp/server-8.0.asc | \  
sudo gpg -o /usr/share/keyrings/mongodb-server-8.0.gpg \  
--dearmor

아래 명령어를 사용하여 MongoDB 저장소를 추가합니다.

echo "deb [ arch=amd64,arm64 signed-by=/usr/share/keyrings/mongodb-server-8.0.gpg ] https://repo.mongodb.org/apt/ubuntu noble/mongodb-org/8.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-8.0.list

저장소가 추가된 후, 다음 명령어를 실행하여 패키지 인덱스를 새로 고치고 MongoDB 서버를 설치합니다. 설치를 확인하려면 ‘Y’를 입력하세요.

sudo apt update && sudo apt install mongodb-org

설치가 완료되면 MongoDB 서비스 ‘mongod’를 시작하고 활성화한 후, MongoDB 서비스 상태를 확인하여 실행 중인지 확인합니다.

sudo systemctl enable --now mongod  
sudo systemctl status mongod

아래에서 MongoDB 서버가 실행 중임을 확인할 수 있습니다.

또한 아래 ‘mongosh’ 명령어로 MongoDB 서버에 로그인할 수 있습니다. 종료하려면 Ctrl+d를 누르세요.

mongosh

Python 및 가상 환경 설정

MongoDB가 설치되었으므로 Python 패키지를 설치하고 프로젝트 디렉토리 및 가상 환경을 설정합니다.

다음 명령어로 Python, Pip 및 Venv 모듈을 설치합니다. 설치를 확인하려면 ‘Y’를 입력하세요.

sudo apt install python3 python3-pip python3-venv

설치가 완료되면 사용자로 로그인합니다.

su - username

이제 새로운 ‘~/app‘ 디렉토리를 생성하고 그 안으로 이동합니다. 이 디렉토리는 FastAPI 프로젝트를 저장하는 데 사용됩니다.

mkdir -p ~/app; cd ~/app

아래 명령어를 실행하여 새로운 ‘venv‘ 가상 환경을 생성하고 활성화합니다. 이로 인해 셸 프롬프트가 ‘(venv) user@hostname‘처럼 됩니다.

python3 -m venv .venv  
source .venv/bin/activate

여기서부터는 작업 환경이 ‘venv’ 가상 환경이어야 합니다. 아래 명령어를 사용하여 ‘venv’에서 로그아웃할 수 있습니다.

deactivate

FastAPI 프로젝트 생성

이제 Python 가상 환경을 생성하고 활성화했으므로 FastAPI를 설치하고 프로젝트 구조를 만듭니다.

‘pip3’ 명령어를 사용하여 ‘fastapi’ 및 ‘uvicorn’ 패키지를 설치합니다.

pip3 install fastapi uvicorn

• ‘fastapi’는 Python에서 API를 구축하기 위한 주요 FastAPI 웹 프레임워크입니다.
• ‘uvicorn’은 Python에서 ASGI(비동기 서버 게이트웨이 인터페이스) 웹 서버의 구현입니다.

설치가 완료되면 다음 명령어로 새로운 파일 및 디렉토리를 생성합니다.

mkdir -p server/{models,routes}  
touch main.py server/{app.py,database.py} server/models/itemModels.py server/routes/item.py

아래는 우리의 FastAPI 프로젝트 구조입니다.

server/app.py

이제 프로젝트가 준비되었으므로 FastAPI 프로젝트의 주요 애플리케이션인 ‘server/app.py’ 파일을 수정합시다.

텍스트 편집기로 ‘app.py‘ 파일을 열고 다음 스크립트를 복사합니다.

from fastapi import FastAPI  
  
app = FastAPI()  
  
@app.get("/")  
async def root():  
 return {"message": "Hello FastAPI!"}

• FastAPI 모듈을 프로젝트에 가져오고 ‘app’ 변수에 바인딩합니다.
• ‘Hello FastAPI!’를 반환하는 새로운 ‘root’ 함수를 생성합니다.
• ‘root’ 함수는 루트 URL에서 GET 메서드에 응답합니다.
• ‘async’는 함수가 비동기 함수임을 표시하며 호출 시 본문 내에서 ‘await’를 사용할 수 있습니다.

main.py

이 섹션에서는 FastAPI 프로젝트를 ‘uvicorn’(Python의 ASGI 웹 서버)을 통해 실행하는 데 사용될 ‘main.py’ 파일을 수정합니다.

이제 ‘main.py‘ 스크립트를 열고 다음 코드를 삽입합니다.

import uvicorn  
  
if __name__ == "__main__":  
 uvicorn.run("server.app:app", host="0.0.0.0", port=8080, reload=True)

• ‘uvicorn’ 모듈을 가져옵니다.
• ‘main.py’ 스크립트가 실행될 때 ‘server/app.py’ 내의 ‘app’ 또는 FastAPI 모듈을 로드합니다.
• FastAPI는 ‘0.0.0.0’에서 포트 ‘8080’으로 실행됩니다.
• 코드 변경 시 자동 재로드를 활성화합니다(‘reload=True’).

FastAPI 프로젝트 실행

이제 프로젝트가 준비되었으므로 첫 번째 FastAPI 프로젝트를 실행해 보겠습니다.

다음과 같이 ‘main.py’ 스크립트를 실행하면 FastAPI가 시스템에서 실행됩니다.

python3 main.py

이제 웹 브라우저를 열고 http://SERVERIP:8080/를 방문하세요. 설치가 성공적이라면 ‘Hello FastAPI!’ 메시지를 볼 수 있습니다. 터미널에서 ‘curl’을 통해서도 접근할 수 있습니다.

마지막으로 http://SERVERIP:8080/docs에서 Swagger UI에서 제공하는 API 문서에 접근할 수 있습니다.

FastAPI와 MongoDB 연결

이 가이드에서는 FastAPI와 MongoDB로 기본 API를 생성합니다. API는 MongoDB 데이터베이스 서버와 CRUD 작업을 수행할 수 있어야 합니다. 이 단계에서는 프로젝트를 MongoDB 서버에 연결합니다.

먼저, 아래 ‘pip3’ 명령어를 실행하여 프로젝트에 ‘motor‘ MongoDB 드라이버를 설치합니다. ‘motor’는 MongoDB 서버에 대한 비차단 및 코루틴 기반 API 액세스를 제공합니다.

pip3 install motor

server/database.py

‘motor‘ 모듈이 설치된 후, ‘server/database.py’ 스크립트를 수정합시다.

텍스트 편집기로 ‘server/database.py’ 파일을 열고 다음 스크립트를 입력합니다. 이는 ‘motor’ 모듈을 통해 MongoDB 서버에 연결하는 데 사용됩니다.

from motor.motor_asyncio import AsyncIOMotorClient  
  
MONGODB_HOST = "mongodb://localhost:27017"  
  
connection = AsyncIOMotorClient(MONGODB_HOST)  
  
database = connection.items  
item_collection = database.get_collection("item_collection")

• ‘motor.motor_asyncio’에서 ‘AsyncIOMotorClient’를 가져옵니다.
• 새로운 상수 ‘MONGODB_HOST’를 생성하고 MongoDB 서버 ‘mongodb://localhost:27017’를 가리킵니다.
• ‘connection’ 변수를 통해 MongoDB 서버에 연결합니다.
• ‘database’ 변수를 통해 ‘items’ 데이터베이스에 연결합니다.
• ‘item_collection’ 변수를 통해 데이터베이스의 컬렉션에 접근합니다.

pydantic으로 데이터베이스 모델 생성

이 섹션에서는 ‘pydantic’을 통해 데이터를 설계합니다. ‘pydantic’은 MongoDB 데이터베이스를 위한 모델링을 제공합니다.

아래 ‘pip3’ 명령어로 ‘pydantic’ 모듈을 설치합니다. ‘pydantic‘ 모듈은 모델을 통해 데이터베이스 스키마를 생성할 수 있는 데이터 검증 라이브러리입니다.

pip3 install pydantic

이제 텍스트 편집기로 ‘server/models/itemModels.py‘ 파일을 열고 다음 스크립트를 복사합니다.

from pydantic import BaseModel, Field  
from typing import Optional  
  
class Item(BaseModel):  
 name: str  
 category: str  
 stocks: int  
 price: int = Field(gt=0)  
  
 class Config:  
 json_schema_extra = {  
 "example": {  
 "name": "Company Smart Watch",  
 "category": "smartwatch",  
 "stocks": 10,  
 "price": 1000,  
 }  
 }  
  
class ItemUpdate(BaseModel):  
 name: Optional[str] = None  
 category: Optional[str] = None  
 stocks: Optional[int] = None  
 price: Optional[int] = None  
  
 class Config:  
 json_schema_extra = {  
 "example": {  
 "name": "New Smart watch",  
 "category": "new-smartwatch",  
 "stocks": 5,  
 "price": 500,  
 }  
 }

• ‘pydantic’에서 ‘BaseModel’ 및 ‘Field’ 모듈을 가져옵니다.
• ‘typing’에서 ‘Optional’ 모듈을 가져옵니다.
• FastAPI를 위한 다음 ‘Item’ 데이터베이스 스키마를 생성합니다: - ‘name’과 ‘category’는 문자열 유형입니다.
• ‘stocks’와 ‘price’는 정수 유형입니다.
• ‘price’는 0보다 커야 합니다.
• ‘Config’ 클래스를 통해 데이터 모델을 확장하여 사용자가 요청에 포함할 수 있는 데이터의 예를 제공합니다.
• 모든 필드가 선택적인 ‘ItemUpdate’ 스키마를 생성합니다.

CRUD 작업 추가

이제 MongoDB 서버에 대한 연결을 생성하고 ‘pydantic’을 통해 데이터베이스 스키마를 생성했습니다. 이제 FastAPI와 MongoDB로 CRUD 작업을 생성해 보겠습니다.

‘server/database.py‘ 스크립트를 다시 열고 ‘ item_helper ‘라는 타입이 ‘ dict ‘인 새로운 함수를 생성합니다. 이 함수를 호출하여 항목에 대한 세부 데이터를 가져옵니다.

def item_helper(item) -> dict:  
 return {  
 "id": str(item["_id"]),  
 "name": item["name"],  
 "category": item["category"],  
 "stocks": item["stocks"],  
 "price": item["price"],  
 }

다음 줄을 추가하여 ‘bson.objectid’에서 ‘ ObjectId ‘ 모듈을 가져옵니다. ‘ ObjectId ‘는 BSON의 데이터 유형의 일부로, MongoDB가 데이터 저장 및 JSON 데이터 표현에 사용합니다.

from bson.objectid import ObjectId

항목 생성

먼저, MongoDB 데이터베이스에 새 데이터를 추가할 수 있는 비동기 함수를 생성합니다.

다음과 같이 새로운 ‘ add_item ‘ 함수를 생성합니다:

# Add a new item  
async def add_item(item_details: dict) -> dict :  
 item = await item_collection.insert_one(item_details)  
 new_item = await item_collection.find_one({"_id": item.inserted_id})  
 return item_helper(new_item)

• ‘add_item’ 함수는 항목 세부 정보의 사전 데이터를 수락합니다.
• ‘item_collection’을 통해 데이터베이스의 문서에 접근하고 ‘insert_one’ 쿼리를 실행하여 새 항목을 추가합니다.
• 새 항목은 성공적인 작업 후 출력됩니다.

모든 항목 검색

둘째, 데이터베이스에서 사용 가능한 모든 항목을 검색할 수 있는 새로운 함수 ‘get_items’를 생성합니다.

MongoDB에서 모든 항목을 검색하기 위한 새로운 ‘ get_items ‘ 함수를 생성합니다.

# retrieve all items  
async def get_items():  
 items = []  
 async for item in item_collection.find():  
 items.append(item_helper(item))  
 return items

• ‘items’의 빈 목록을 생성합니다.
• ‘find()’ 쿼리를 사용하여 모든 데이터를 검색하고 이를 반복합니다.
• 각 항목은 ‘append’ 메서드를 통해 ‘items’ 목록에 추가됩니다.
• 루프가 완료되면 항목이 표시됩니다.

특정 ID에 따라 항목 검색

다음으로, 특정 선택자 ‘id’에 대한 데이터를 검색할 수 있는 새로운 함수를 생성합니다. 이는 특정 항목에 대한 세부 데이터를 보여줍니다.

특정 항목의 데이터를 검색하기 위한 새로운 ‘ get_item ‘ 함수를 생성합니다. ‘get_item’ 함수는 선택자로서 문자열 ‘id’를 수락하고 사전을 반환합니다.

# retrieve specific item  
async def get_item(id: str) -> dict:  
 item = await item_collection.find_one({"_id": ObjectId(id)})  
 if item:  
 return item_helper(item)  
 return "Item Not Found."

• ‘find_one()’ 쿼리를 사용하여 ‘id’에 따라 데이터를 검색합니다.
• 항목이 발견되면 ‘item_helper’ 사전 형식을 사용하여 세부 정보가 표시됩니다.
• 항목이 없으면 응답은 ‘Item Not Found’입니다.

항목 업데이트

이제 항목을 업데이트하기 위한 새로운 함수를 생성합니다. API를 통해 항목을 부분적으로 업데이트할 수도 있습니다.

다음 코드를 사용하여 새로운 ‘ change_item ‘ 함수를 정의합니다:

# update item  
async def change_item(id: str, data: dict):  
 if len(data) < 1:  
 return "Please input your data"  
 find_item = await item_collection.find_one({"_id": ObjectId(id)})  
 if find_item:  
 item_update = await item_collection.update_one({"_id": ObjectId(id)}, {"$set": data})  
 if item_update:  
 return True  
 return False

• ‘change_item’ 함수는 두 개의 인수, 즉 대상 항목의 ‘id’와 새로운 데이터인 ‘data’를 받습니다.
• 데이터가 비어 있거나 ‘< 1’인 경우 작업이 종료됩니다.
• 이 함수는 ‘id’ 선택자에 따라 항목을 찾습니다.
• ‘id’가 발견되면 ‘item_update’가 실행됩니다.
• ‘item_update’가 성공하면 ‘True’를 반환하고, 그렇지 않으면 ‘False’를 반환합니다.

항목 삭제

마지막으로, 특정 선택자를 통해 항목을 삭제하기 위한 ‘delete_item’ 함수를 생성합니다.

항목을 삭제하기 위해 ‘ delete_item ‘ 함수를 생성하고 ‘id’ 선택자를 사용합니다:

# delete an item  
async def delete_item(id: str):  
 item = await item_collection.find_one({"_id": ObjectId(id)})  
 if item:  
 await item_collection.delete_one({"_id": ObjectId(id)})  
 return(f'Item {id} deleted.')  
 return "Item Not Found."

• ‘find_one()’ 쿼리는 제공된 ‘id’에 따라 항목을 검색합니다.
• 항목이 발견되면 ‘delete_one()’ 쿼리를 통해 삭제되고 ‘Item id deleted’가 반환됩니다.
• 항목이 없으면 ‘Item Not Found’가 표시됩니다.

CRUD 작업을 위한 라우트 추가

이 시점에서 FastAPI로 CRUD 작업을 위한 비동기 함수를 생성했습니다. 이제 각 작업에 대한 라우트 또는 엔드포인트를 생성해 보겠습니다.

‘server/routes/item.py‘ 파일을 선호하는 편집기로 편집합니다.

server/routes/item.py

먼저, ‘fastapi’에서 ‘ APIRouter ‘ 및 ‘ Body ‘ 모듈을 추가합니다. 그런 다음 ‘ jsonable_encode ‘를 ‘fastapi.encoders’에서 추가합니다.

from fastapi import APIRouter, Body  
from fastapi.encoders import jsonable_encoder

‘database.py‘ 파일에서 각 함수를 가져옵니다.

from server.database import (  
 add_item,  
 get_items,  
 get_item,  
 change_item,  
 delete_item,  
)

‘itemModels.py‘ 파일에서 ‘Item’ 및 ‘ItemUpdate’ 모델을 가져옵니다.

from server.models.itemModels import (  
 Item,  
 ItemUpdate,  
)

‘APIRouter’ 클래스를 ‘router’ 변수로 호출합니다.

router = APIRouter()

새로운 항목 추가를 위한 라우트

이제 새로운 항목을 추가하기 위한 라우트를 추가합시다. 이 예제에서는 ‘/item‘ URL에 POST를 통해 새로운 항목을 추가할 수 있습니다.

새로운 항목을 추가하기 위한 라우트를 설정하기 위해 다음 줄을 추가합니다. 항목의 루트 URL에 대한 모든 POST 요청은 새로운 데이터를 삽입하는 것으로 처리됩니다.

# add new item operation  
@router.post("/")  
async def add_item_data(item: Item = Body(...)):  
 item = jsonable_encoder(item)  
 new_item = await add_item(item)  
 return new_item

• ‘add_item_data’ 함수는 요청의 ‘Body’의 일부가 될 ‘Item’ 스키마를 수락합니다.
• 항목은 ‘jsonable_encoder’를 통해 사전 형식으로 변환됩니다.
• ‘add_item’ 함수를 통해 사전 데이터를 삽입합니다(데이터베이스.py 참조) ‘new_item’ 변수 위에.
• 삽입된 데이터 ‘new_item’을 응답으로 반환합니다.

모든 항목을 가져오는 라우트

데이터를 검색하기 위한 라우트를 설정하기 위해 다음 스크립트를 추가합니다. 항목의 루트 URL에 대한 모든 GET 요청은 모든 데이터를 검색합니다.

# get all available items  
@router.get("/")  
async def get_item_data():  
 items = await get_items()  
 if items:  
 return items  
 return "No available item."

• ‘get_item_data’ 함수를 생성하여 ‘database.py’ 파일의 ‘get_item’ 함수를 실행합니다.
• 항목이 사용 가능한 경우, 모든 항목의 목록을 가져옵니다.
• 항목이 없으면 ‘No available item’ 응답을 받습니다.

특정 항목을 가져오는 라우트

특정 항목의 세부 정보를 가져오기 위해 ‘id’를 선택자로 사용할 것입니다. ‘/id’에 대한 모든 GET 요청은 요청된 ‘id’의 세부 항목을 반환합니다.

# Show details of the item via the id  
@router.get("/{id}")  
async def get_item_details(id):  
 item_details = await get_item(id)  
 if item_details:  
 return item_details  
 return "Item not found."

• ‘get_item_details’ 함수를 생성하고 URL에서 ‘id’를 전달합니다.
• ‘get_item’ 함수(데이터베이스.py 참조)를 호출하고 ‘id’를 인수로 전달합니다.
• 항목이 발견되면 항목의 세부 정보가 표시됩니다.
• 특정 ‘id’의 항목이 없으면 ‘Item not found’를 받습니다.

항목 업데이트를 위한 라우트

업데이트된 항목을 위한 라우트를 설정하기 위해 다음 코드를 복사합니다:

# Update Item  
@router.put("/{id}")  
async def update_item(id: str, data: ItemUpdate = Body(...)):  
 data = {k: v for k, v in data.dict().items() if v is not None}  
 updated_item = await change_item(id, data)  
 if updated_item:  
 return{f'Success: item {id} updated.'}  
 return "Error"

• ‘update_item’ 함수는 두 개의 인수, ‘id’ 선택자와 ‘ItemUpdate’ 모델을 기반으로 한 ‘data’를 받습니다.
• ‘data’ 변수에서 데이터를 확인합니다.
• ‘updated_item’은 ‘database.py’ 파일의 ‘change_item’ 함수를 실행합니다.
• 업데이트가 성공하면 ‘Success’라는 출력을 볼 수 있습니다.

항목 삭제를 위한 라우트

항목 삭제를 위한 ‘remove_item’ 함수를 생성하기 위해 다음 줄을 삽입합니다. 특정 ‘/id’에 대한 모든 DELETE 작업은 ‘id’와 일치하는 항목을 제거합니다.

# delete item via id  
@router.delete("/{id}")  
async def remove_item(id):  
 item_to_delete = await delete_item(id)  
 if item_to_delete:  
 return item_to_delete  
 return{f'Item {id} Not Available.'}

• ‘remove_item’ 함수는 ‘delete_item’을 실행하고 ‘id’ 선택자를 전달합니다.
• 삭제 작업은 ‘item_to_delete’ 변수에 저장되고 실행됩니다.
• 항목이 사용 가능하지 않으면 ‘Item id Not Available’을 받습니다.

server/app.py

이제 ‘server/routes/item.py’ 파일을 완료했으므로 ‘server/app.py’에 포함시킵니다.

텍스트 편집기로 ‘app.py‘ 파일을 열어 ‘ server/routes/item.py ‘ 파일에서 ‘router’를 ‘ ItemRouter ‘로 가져옵니다.

from server.routes.item import router as ItemRouter

기본 접두사 ‘/item‘으로 ‘ItemRouter’를 삽입합니다. CRUD 작업은 ‘/item‘ URL을 통해 처리됩니다.

app.include_router(ItemRouter, tags=["Item"], prefix="/item")

이제 CRUD 엔드포인트는 다음에서 사용할 수 있습니다:

• 새로운 항목 추가: POST to ‘/item‘
• 모든 항목 검색: GET to ‘/item‘
• 특정 항목 검색: GET to ‘/item/id’. ‘id’는 MongoDB에 의해 생성됩니다.
• 항목 업데이트: PUT to ‘/item/id’
• 항목 삭제: DELETE to ‘/item/id’

CRUD 작업 테스트

먼저 FastAPI 프로젝트가 실행 중인지 확인하거나 다음과 같이 ‘main.py’ 스크립트를 실행합니다:

python3 main.py

http://SERVERIP:8080/docs로 이동하면 생성한 각 라우트를 볼 수 있습니다.

아래는 새로운 항목을 추가하는 예입니다.

API를 통해 모든 사용 가능한 항목을 검색합니다.

‘id’ 선택자를 통해 특정 항목을 검색합니다.

특정 항목에서 부분 데이터를 업데이트합니다.

아래는 업데이트된 데이터입니다.

아래는 ‘id’ 선택자를 통해 삭제 작업입니다.

결론

축하합니다! Ubuntu 24.04에서 MongoDB와 함께 FastAPI 설치를 완료했습니다. 또한 ‘motor’ 모듈을 사용하여 FastAPI를 MongoDB에 연결하는 방법, ‘pydantic’을 통해 데이터 모델을 생성하는 방법, FastAPI로 CRUD 작업을 생성하는 방법, API의 엔드포인트를 생성하는 방법을 배웠습니다.