PythonでMongoDBのODMを使用しよう

前書

この記事はPythonでMongoDBを入門しようという記事の続編になりますが、
前回の記事を読んでいなくても内容がわかる構成にしています。
もしMongoDBに興味があれば、最後まで付き合っていただけると幸いです。

環境構築

MongoDB

この記事ではDockerを使用して環境構築行います。
ローカル内で直接MongoDBを入れたい方は前回の記事を参考にしてください。

docker-compose.yml

version: '3.1'

services:

  mongo:
    image: mongo
    restart: always
    ports:
      - 27017:27017      
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: example

  mongo-express:
    image: mongo-express
    restart: always
    ports:
      - 8081:8081
    environment:
      ME_CONFIG_MONGODB_ADMINUSERNAME: root
      ME_CONFIG_MONGODB_ADMINPASSWORD: example

コンテナを立ち上げます。

docker-compose up -d mongo mongo-express

Mongo Expressにアクセス、表示出来たらMongoDBの環境構築は完了です。

Python

Python側は mongoengineというライブラリを使用します。

pip install mongoengine

試しに接続します。
デフォルトで入ってるlocalデータベースに接続してみます。

test.py

from mongoengine import connect

connect(db='local',
        username="root",
        password="example",
        host='192.168.99.100',
        port=27017,
        authentication_mechanism='SCRAM-SHA-1',
        authentication_source='admin'
        )

実行し、エラーが出なければ成功です。
その他の接続方法はこちらを参考にしてくださいPythonでMongoDBを入門しよう

MongoDBのODM

ODMとは何か、文字通りで言うとオブジェクトドキュメントマッピングになります。
MongoDBのようなドキュメントデータベースにオブジェクトをマッピングします。
使用するメリットとしてはデータの構造を縛れることです。

実例を見ていきます

test.py

from mongoengine import connect, Document, EmbeddedDocument, \
    StringField, IntField, DateTimeField, ListField, EmbeddedDocumentField
from datetime import datetime

connect(db='company',
        username="root",
        password="example",
        host='192.168.99.100',
        port=27017,
        authentication_mechanism='SCRAM-SHA-1',
        authentication_source='admin'
        )


class Employee(EmbeddedDocument):
    """
        社員詳細
    """
    name = StringField(required=True)
    age = IntField(required=False)


SCALE_CHOICES = (
    ("venture", "ベンチャー"),
    ("major", "大手")
)


class Company(Document):
    """
        会社モデル
    """
    name = StringField(required=True, max_length=32)
    scale = StringField(required=True, choices=SCALE_CHOICES)
    created_at = DateTimeField(default=datetime.now())
    members = ListField(EmbeddedDocumentField(Employee))

Djangoやsqlalchemy 使用してモデルを定義したことがあれば、馴染む構成になっています。
実際に定義されたモデルを使用してデータをMongoDBに入れてみます。
下記のコードをtest.pyに追記し、実行します。
テスト用のデータベースをあらかじめ作る必要はありません。

test.py

class TestMongoEngine:
    def add_one(self):
       c_obj = Company(
           name="有名ベンチャー",
           scale="venture",
       )
       c_obj.save()
       return c_obj


if __name__ == "__main__":
        t = TestMongoEngine()
        t.add_one()

Mongo Expressから、データが入れられたことが確認できます。

EmbeddedDocumentFieldはモデルの内部に更に構造を定義したい場合に、使用すると役に立つので使ってみます。

先ほどtest.pyに追記されたコードを以下のように修正し、再度実行します。

test.py

class TestMongoEngine:
    def add_one(self):
       member1 = Employee(
               name="memberA",
               age=40,
       )
       member2 = Employee(
               name="memberB",
               age=35,
       )
       c_obj = Company(
           name="有名ベンチャーA",
           scale="venture",
           members=[member1, member2]
       )
       c_obj.save()
       return c_obj


if __name__ == "__main__":
        t = TestMongoEngine()
        t.add_one()

Mongo Expressから結果を確認します。

ODM使用してCRUD

C(Create)新規追加する例は最初の例で紹介したので、データの読み込みから紹介していきます。

R(Retrieve)データの読み込み

単一読み込み

先程新規追加したデータベースから一つのデータを読み込みます。

test.py

...
class TestMongoEngine: 
    def get_one(self):
        return Company.objects.first()

if __name__ == "__main__":
        t = TestMongoEngine()
        rest = t.get_one()
        print(rest.id)
        print(rest.name)

結果

5e7ed47419d1a75baa2bc3f3
有名ベンチャー

全部読み込み

test.py

...
class TestMongoEngine:
    def get_more(self):
        return Company.objects.all()

if __name__ == "__main__":
        t = TestMongoEngine()
        rest = t.get_more()
        print(rest)

結果

[<Company: Company object>, <Company: Company object>, <Company: Company object>]

id検索で読み込み

test.py

...
class TestMongoEngine:
      def get_from_oid(self, oid):
          return Company.objects.filter(pk=oid).first()

if __name__ == "__main__":
        t = TestMongoEngine()
        rest = t.get_from_oid("5e7ed47419d1a75baa2bc3f3")
        print(rest.id)
        print(rest.name)

結果

5e7ed47419d1a75baa2bc3f3
有名ベンチャー

補足

データ読み込みする際に順番を並び替えたい場合、 metaを会社モデルに追記します。

test.py

class Company(Document):
    """
        会社モデル
    """
    name = StringField(required=True, max_length=32)
    scale = StringField(required=True, choices=SCALE_CHOICES)
    created_at = DateTimeField(default=datetime.now())
    members = ListField(EmbeddedDocumentField(Employee))
    meta = {
        'ordering': ['-created_at']  # metaを追記
    }

ドキュメント

データの更新(Update)

単一のデータ修正

test.py

...
class TestMongoEngine:
    def update(self):
        rest = Company.objects.filter(name="有名ベンチャー").update_one(name="普通のベンチャー")
        return rest

if __name__ == "__main__":
        t = TestMongoEngine()
        rests = t.update()

実行後、Mongo Expressからデータ変更されたことが確認できます。

複数のデータ修正

MongoDBにnameが有名ベンチャーAになってる記録が二つ存在します、それらを修正していきます。

test.py

class TestMongoEngine:
    def update(self):
        rest = Company.objects.filter(name="有名ベンチャーA").update(name="有名ベンチャーB")
        return rest

if __name__ == "__main__":
        t = TestMongoEngine()
        rests = t.update()
        print(rests)

実行結果

2

Mongo Expressからもデータ修正されたことが確認できます。

データの削除(Delete)

単一のデータ削除

test.py

class TestMongoEngine:
    def delete(self):
        rest = Company.objects.filter(name="普通のベンチャー").first().delete()
        return rest

if __name__ == "__main__":
        t = TestMongoEngine()
        rests = t.delete()
        print(rests)

実行後

複数のデータ削除

nameが有名ベンチャーBになってる二件のデータを削除します。

test.py

...
class TestMongoEngine:
    def delete(self):
        rest = Company.objects.filter(name="有名ベンチャーB").delete()
        return rest

if __name__ == "__main__":
        t = TestMongoEngine()
        rests = t.delete()
        print(rests)

実行結果

2

Mongo Expressから確認すると、CompanyDBに入ってるデータすべてが削除されました。

後書

MongoEngineのドキュメント
今度時間あればFlaskとMongoDB使用してWebサービス作る記事を書きたいと思います。