API 서비스 개발(2) - DB

개요

이번엔 Fast API 서비스와 Database를 연동하는 작업을 해보겠다.

 

데이터베이스 설치

개발하려는 서비스 요건상 JSON 포맷을 써야하는 데이터가 많기 때문에 MongoDB를 설치하자.

 

1. 다운로드

Download MongoDB Community Server | MongoDB

Try MongoDB Community Edition

위 링크에 접속해서 .msi 파일을 받는다.

 

2. 환경 설정

특별한 변경 없이 위와같이 설치하며 Data, Log의 경로를 기억해두자

 

MongoDB Compass는 아래와같이 사용자가 GUI를 통해 편리하게 관리하기 위한 툴이다.

 

기본 주소 mongodb://localhost:27017 를 기억해두자

 

 

데이터베이스 연결

1. 라이브러리 선택

Python과 MongoDB를 연결하기 위한 라이브러리는 두 종류가 있다.

 

- PyMongo: PyMongo는 동기(Synchronous) MongoDB 드라이버

- Motor: Motor는 비동기(Asynchronous) MongoDB 드라이버

 

이번에는 굳이 데이터 저장에 비동기를 써가며 디버깅과 모니터링같은 관리의 복잡성이 필요하지 않기 때문에 리스크를 줄이기 위해 동기화 방식 PyMongo을 사용한다.

 

2. 라이브러리 설치

pip install pymongo

 

3. DB 연결 구현 (set_database.py)

from pymongo import MongoClient

# MongoDB 클라이언트 설정
client = MongoClient("mongodb://localhost:27017")
db = client.test_db  # 사용할 데이터베이스 이름
collection = db.test_collection  # 사용할 컬렉션 이름

 

4. DB 객체 호출

from fastapi import FastAPI
from database.set_database import db

app = FastAPI()

@app.get("/test")
async def test():
    test_collection = db.test_collection
    print("test_collection : ", test_collection)
    result = test_collection.find_one()
    return {result}

 

출력 결과는 아래와 같다.

test_collection :  Collection(Database(MongoClient(host=['localhost:27017'], document_class=dict, tz_aware=False, connect=True), 'test_db'), 'test_collection')

 

 

 

데이터베이스 인터페이스

1. 데이터 삽입

간단하게 아래 코드로 데이터를 넣어보자.

@app.post("/items/", response_model = ItemInDB)
def create_item(item: Item):
    item_dict = item.dict()
    result = test_collection.insert_one(item_dict)
    item_in_db = ItemInDB(id=str(result.inserted_id), **item_dict)
    return item_in_db

 

Swagger로 POST /items/ 를 호출해서 데이터가 잘 들어갔는지 확인해 볼 수 있다.

 

 

2. 데이터 조회

우선 데이터는 MongoDB Compress 에서도 확인해볼 수 있다.

 

또는 API를 짜서 확인해볼 수 있다.

# 데이터 조회
@app.get("/items/", response_model=List[ItemInDB])
def get_items():
    items = list(test_collection.find())
    if not items:
        return []
    items_in_db = []
    for item in items:
        item_id = str(item["_id"]["$oid"]) if isinstance(item["_id"], dict) and "$oid" in item["_id"] else str(item["_id"])
        items_in_db.append(ItemInDB(id=item_id, **item))
    return items_in_db

 

swagger에서 결과를 확인해보자.

 

 

 

감사합니다.