python을 이용해 async로 mongodb 데이터 조회하기

환경구성

MongoDB

docker-compose.yaml 파일을 작성합니다.

version: '3.8'
services:
  mongodb:
    image: mongo
    container_name: mongodb
    restart: always
    ports:
      - 27017:27017
    volumes:
      - ./mongodb:/data/db
    environment:
      - MONGO_INITDB_ROOT_USERNAME=root
      - MONGO_INITDB_ROOT_PASSWORD=1234 
      - MONGO_INITDB_DATABASE=mydb

docker로 local mongodb 실행합니다.

docker-compose up -d

Python

python package pymongo 설치합니다.

pip install pymongo

pymongo DB connection

import pymongo

client = pymongo.MongoClient(
    host="0.0.0.0",
    port=27017,
    username="root",
    password="1234",
)
db = client.get_database("mydb")
collection = db.get_collection("test")

data 조회합니다.

items = collection.find()
[item for item in items]  # 결과 []

Insert Data

데이터 조회를 위해 랜덤 데이터를 생성해서 적재합니다.

import datetime
import random

n = 1000000
names = ["mongo", "python", "docker", "async"]
actions = ["create", "read", "update", "delete"]

base_timestamp = datetime.datetime(2023, 7, 1, 13, 52, 0)
timegaps = list(range(n))
random.shuffle(timegaps)

docs = []
for timegap in timegaps:
    doc = {
        "name": random.choice(names),
        "action": random.choice(actions),
        "value": random.randint(1, 30),
        "timestamp": base_timestamp + datetime.timedelta(seconds=timegap)
    }
    docs += [doc]

results = collection.insert_many(docs)

collection.find_one()
# {'_id': ObjectId('64c50f6340cd550aa4275422'),
#  'name': 'async',
#  'action': 'update',
#  'value': 25,
#  'timestamp': datetime.datetime(2023, 7, 6, 0, 55, 18)}

Find Data

sync/async 데이터 조회를 비교합니다.

sync

(sync) mongo cliect를 선업합니다.

import pymongo

client = pymongo.MongoClient(
    host="0.0.0.0", port=27017, username="root", password="1234"
)
db = client.get_database("mydb")
collection = db.get_collection("test")

name 기반으로 데이터를 조회하는 find_by_name 함수를 작성합니다.

def find_by_name(name):
    items = collection.find(
        {"name": name}, sort=[("timestamp", -1)]
    )
    return list(items)

name 을 list 로 받아 데이터를 조회합니다.

def main(names):
    started_at = time.time()
    for name in names:
        items = find_by_name(name)
    print(f"finished: {time.time() - started_at}")

main(["aync"])
# finished: 1.4455139636993408
main(["aync", "python", "mongo"])
# finished: 4.6754419803619385

async

python에서 mongo async를 지원하는 package motor 설치합니다.

pip install motor

주피터 노트북 환경에서 async 실행시 아래 코드를 먼저 실행합니다.

import nest_asyncio
nest_asyncio.apply()

(async) mongo cliect를 선업합니다.

from motor.motor_asyncio import AsyncIOMotorClient

async_client = AsyncIOMotorClient("mongodb://root:1234@0.0.0.0:27017")
async_db = async_client.get_database("mydb")
async_collection = async_db.get_collection("test")

name 기반으로 데이터를 조회하는 find_by_name 함수를 작성합니다. 아래 두가지 방식 가능

async def find_by_name_async(name):
    cursor = async_collection.find(
        {"name": name}, sort=[("timestamp", -1)]
    )
    docs = []
    async for doc in cursor:
        docs += [doc]
    return docs

async def find_by_name_async(name):
    cursor = async_collection.find(
        {"name": name}, sort=[("timestamp", -1)]
    )
    docs = []
    for doc in await cursor.to_list(length=None):
        docs += [doc]
    return docs

name 을 list 로 받아 asyncio 방식으로 데이터를 조회합니다.

import asyncio

async def main_async(names):
    started_at = time.time()
    await asyncio.gather(
        *[find_by_name_async(name) for name in names]
    )
    print(f"finished: {time.time() - started_at}")

await main_async(["async"])
# finished: 2.0516769886016846
await main_async(["async", "mongo", "python"])
# finished: 4.005278825759888

pymongo find 동작 방식으로는 sync/async 조회 시간에 차이가 없어 보입니다.
추가 내용을 다음에 확인해보겠습니다.