BEANIE Archives - icodebroker

[PYTHON/BEANIE] Document 클래스 : Collection 서브 클래스와 name 변수를 사용해 MongoDB 컬렉션명 설정하기

■ Document 클래스에서 Collection 서브 클래스와 name 변수를 사용해 MongoDB 컬렉션명을 설정하는 방법을 보여준다. ▶ 예제 코드 (PY)


from typing import Optional, List
from beanie import Document

class Event(Document):
    creator     : Optional[str]
    title       : str
    image       : str
    description : str
    tagList     : List[str]
    location    : str

    class Collection:
        name = "event"

from typing import Optional, List

from beanie import Document

class Event(Document):

creator : Optional[str]

title : str

image : str

description : str

tagList : List[str]

location : str

class Collection:

name = "event"

[PYTHON/BEANIE] Document 클래스 : MongoDB 데이터베이스에서 단순 CRUD 애플리케이션 만들기

■ Document 클래스를 사용해 MongoDB 데이터베이스에서 단순 CRUD 애플리케이션을 만드는 방법을 보여준다. ▶ event.py


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]

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


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)

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


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()

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()

▶