■ Document 클래스를 사용해 MongoDB 데이터베이스에서 단순 CRUD 애플리케이션을 만드는 방법을 보여준다.
▶ event.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
from typing import Optional, List from pydantic import BaseModel from beanie import Document class Event(Document): title : str image : str description : str tagList : List[str] location : str class Settings: name = "event" class EventUpdate(BaseModel): title : Optional[str] image : Optional[str] description : Optional[str] tagList : Optional[List[str]] location : Optional[str] |
▶ db_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
from motor.motor_asyncio import AsyncIOMotorClient from beanie import init_beanie class DBHelper: def __init__(self): pass async def initialize(self, databaseURL, documentModelList): self.databaseURL = databaseURL self.documentModelList = documentModelList client = AsyncIOMotorClient(self.databaseURL) await init_beanie(database = client.get_default_database(), document_models = self.documentModelList) |
▶ document_helper.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
from typing import Any, List from beanie import PydanticObjectId from pydantic import BaseModel class DocumentHelper: def __init__(self, modelMetaClass): self.modelMetaClass = modelMetaClass async def create(self, document) -> str: document = await document.create() return document.id async def get(self, id : PydanticObjectId) -> Any: document = await self.modelMetaClass.get(id) if document: return document return False async def getList(self) -> List[Any]: documentList = await self.modelMetaClass.find_all().to_list() return documentList async def update(self, id : PydanticObjectId, updateModel : BaseModel) -> Any: documentID = id updateModelDictionary = updateModel.dict() updateModelDictionary = {k : v for k, v in baseModelDictionary.items() if v is not None} updateQuery = {"$set" : {field: value for field, value in updateModelDictionary.items()}} document = await self.get(documentID) if not document: return False await document.update(updateQuery) return document async def delete(self, id : PydanticObjectId) -> bool: document = await self.get(id) if not document: return False await document.delete() return True async def deleteAll(self): await self.modelMetaClass.find_all().delete() |
▶ main.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
import asyncio from db_helper import DBHelper from document_helper import DocumentHelper from event import Event, EventUpdate async def main(): dbHelper = DBHelper() await dbHelper.initialize("mongodb://localhost:27017/testdb", [Event]) documentHelper = DocumentHelper(Event) await documentHelper.deleteAll() # 이벤트를 5개 생성한다. for i in range(1, 6): event = Event(title = f"제목 {i}", image = f"이미지 {i}", description = f"설명 {i}", tagList = [f"태그 {i}"], location = f"장소 {i}") await documentHelper.create(event) # 이벤트 리스트를 조회한다. eventList = await documentHelper.getList() for event in eventList: print(event) print() # 첫번째 이벤트를 가져온다. event = await documentHelper.get(eventList[0].id) print(event) print() # 첫번째 이벤트를 수정한다. eventUpdate = EventUpdate() eventUpdate.image = "이미지 *" await documentHelper.update(event.id, eventUpdate) # 이벤트 리스트를 조회한다. eventList = await documentHelper.getList() for event in eventList: print(event) print() # 첫번째 이벤트를 가져온다. event = await documentHelper.get(eventList[0].id) # 첫번쨰 이벤트를 삭제한다. await documentHelper.delete(event.id) # 이벤트 리스트를 조회한다. eventList = await documentHelper.getList() for event in eventList: print(event) print() asyncio.run(main()) |
▶ requirements.py
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
anyio==3.5.0 beanie==1.10.4 cffi==1.15.0 click==8.0.4 idna==3.7 motor==2.5.1 multidict==6.0.2 pycparser==2.21 pydantic==1.9.0 pymongo==3.12.3 sniffio==1.2.0 toml==0.10.2 typing_extensions==4.1.1 yarl==1.9.4 |
▶ 테스트 실행 명령
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
(env) king@cosmos:~/sample$ python3 main.py id=ObjectId('66534d179aaefca050f2ae0c') revision_id=UUID('372066f8-a892-417c-8d81-1fc010063380') title='제목 1' image='이미지 1' description='설명 1' tagList=['태그 1'] location='장소 1' id=ObjectId('66534d179aaefca050f2ae0d') revision_id=UUID('6565e4db-efcf-4bda-98e7-444fb2f2a1ab') title='제목 2' image='이미지 2' description='설명 2' tagList=['태그 2'] location='장소 2' id=ObjectId('66534d179aaefca050f2ae0e') revision_id=UUID('4bcc9346-89bb-4c4e-9869-64d6ae54107f') title='제목 3' image='이미지 3' description='설명 3' tagList=['태그 3'] location='장소 3' id=ObjectId('66534d179aaefca050f2ae0f') revision_id=UUID('d0d73d61-f681-40d8-8fe5-74673410a502') title='제목 4' image='이미지 4' description='설명 4' tagList=['태그 4'] location='장소 4' id=ObjectId('66534d179aaefca050f2ae10') revision_id=UUID('f056e2a0-c01c-4765-bffe-16cf8603550d') title='제목 5' image='이미지 5' description='설명 5' tagList=['태그 5'] location='장소 5' id=ObjectId('66534d179aaefca050f2ae0c') revision_id=UUID('1ff3c352-bc90-4668-8097-22ce8c4ced80') title='제목 1' image='이미지 1' description='설명 1' tagList=['태그 1'] location='장소 1' id=ObjectId('66534d179aaefca050f2ae0c') revision_id=UUID('6ff3d121-4e61-4e02-a3b1-9ffa92377164') title='제목 1' image='이미지 *' description='설명 1' tagList=['태그 1'] location='장소 1' id=ObjectId('66534d179aaefca050f2ae0d') revision_id=UUID('e4c909ad-ae31-4010-8902-7d3d3da36caf') title='제목 2' image='이미지 2' description='설명 2' tagList=['태그 2'] location='장소 2' id=ObjectId('66534d179aaefca050f2ae0e') revision_id=UUID('7331035d-cc8b-4483-9932-abb449a58388') title='제목 3' image='이미지 3' description='설명 3' tagList=['태그 3'] location='장소 3' id=ObjectId('66534d179aaefca050f2ae0f') revision_id=UUID('5421957b-4a6f-4ecf-bdbd-502c645e7977') title='제목 4' image='이미지 4' description='설명 4' tagList=['태그 4'] location='장소 4' id=ObjectId('66534d179aaefca050f2ae10') revision_id=UUID('2eff1b0f-fbd1-4867-8801-523de019bcee') title='제목 5' image='이미지 5' description='설명 5' tagList=['태그 5'] location='장소 5' id=ObjectId('66534d179aaefca050f2ae0d') revision_id=UUID('b880393d-b6ca-4919-a264-440c71bdbbce') title='제목 2' image='이미지 2' description='설명 2' tagList=['태그 2'] location='장소 2' id=ObjectId('66534d179aaefca050f2ae0e') revision_id=UUID('b7313d8e-f045-4017-868f-9ad4f664592f') title='제목 3' image='이미지 3' description='설명 3' tagList=['태그 3'] location='장소 3' id=ObjectId('66534d179aaefca050f2ae0f') revision_id=UUID('a00e71a1-3651-4687-bed8-f1af032f4e75') title='제목 4' image='이미지 4' description='설명 4' tagList=['태그 4'] location='장소 4' id=ObjectId('66534d179aaefca050f2ae10') revision_id=UUID('f153887a-ceff-441c-944b-d20010efa710') title='제목 5' image='이미지 5' description='설명 5' tagList=['태그 5'] location='장소 5' |