Python を使用して MongoDB データベースにローカル接続する

PyMongo（MongoDB の Python ドライバー）を使うと、MongoDB のデプロイ先に接続し、データベースとコレクションを選択して、BSON ドキュメントを挿入したり検索したりできます。以下では MongoDB URI と小規模な買い物リスト用コレクションを使います。URI のプレースホルダーは実際のデプロイ先の接続情報に置き換え、認証情報をソース管理に含めないでください。(insert_many(); insert_one())

Python と PyMongo のインストール

まず Python をインストールし、python -m pip install pymongo で PyMongo をインストールします。この例では mongodb+srv:// で始まる DNS シードリスト URI を使うため、DNS サポートも python -m pip install "pymongo[srv]" でインストールしてください（または dnspython を別途インストールします）。デプロイ先から標準の mongodb:// URI が提供されている場合は、その URI を使い、指定されたネットワーク設定と認証設定に従ってください。

プログラムを pymongo_example.py として保存し、python pymongo_example.py でローカルのデモを実行します。データベースを使う処理を実行するには、まず MONGODB_URI に MongoDB のデプロイ先から提供された接続文字列を設定し、--database を指定します。macOS または Linux では export MONGODB_URI='mongodb+srv://<username>:<password>@<cluster-name>/' を、PowerShell では $env:MONGODB_URI = 'mongodb+srv://<username>:<password>@<cluster-name>/' を使います。プレースホルダーを置き換え、URI の形式に応じて認証情報の特殊文字を URL エンコードしてください。また、実際の認証情報をソース管理や共有されるシェル履歴に残さないでください。python pymongo_example.py --database を実行します。保存先のファイル名とコマンド内のファイル名は同じです。

PyMongo で MongoDB に接続する

MongoDB のデプロイ先から提供された URI を使って MongoClient を作成します。シードリスト URI にはクラスターのホスト名が含まれ、認証オプションが含まれる場合もあります。<username>、<password>、<cluster-name> はそのまま使わず、実際の値に置き換えてください。URI に認証情報が含まれることがあるため、出力して表示しないでください。以下のデータベース処理は --database を指定した場合にのみ選択され、MONGODB_URI から URI を読み込みます。このフラグを付けない場合、プログラムは別のオフラインデモを実行し、MongoDB には接続しません。

アプリケーションでは、MONGODB_URI などの環境変数から URI を読み取り、MongoClient に渡します。スクリプトでは、モジュール直下の if __name__ == "__main__": ガード内に処理を置くと、ファイルをインポートしただけでデータベース操作が始まるのを防げます。MongoClient は遅延接続するため、最初のデータベース操作より前にクライアントを作成できます。ネットワーク、DNS、認証、サーバー選択の問題は、最初の操作時に発生することがよくあります。処理が終わったらクライアントを閉じるか、短時間実行するスクリプトではコンテキストマネージャーを使ってください。

データベースとコレクションを作成する

client["user_shopping_list"] でデータベースを、database["items"] でコレクションを選択します。これらの式が返すのはハンドルであり、それだけではデータベースやコレクションのデータが永続化されるわけではありません。MongoDB は、最初の挿入でデータが書き込まれたときに、それらを作成します。挿入と検索の各手順では、一貫して同じ名前を使ってください。

ドキュメントを挿入する

Python の辞書は MongoDB のドキュメントにそのまま対応します。データベースを使う例では、2件のドキュメントに collection.insert_many(documents) を使い、3件目には collection.insert_one(document) を使います。3件目にはタイムゾーン対応の Python datetime.datetime 値が含まれ、PyMongo がこれを BSON の日付としてエンコードします。戻り値の結果オブジェクトには、挿入したドキュメントを検索するときに使える識別子が含まれます。MongoDB に _id を自動生成させる場合は _id を省略してください。この例では、検索対象を挿入したレコードに限定できるよう、一意の ID を割り当てています。

挿入処理には、接続可能な MongoDB のデプロイ先と書き込み権限が必要です。重複レコードを許容するか、例で安定した識別子を使って重複キーエラーを処理している場合を除き、共有データベースに対してサンプルの挿入を繰り返さないでください。

ドキュメントを検索し、フィールドがない場合に対処する

collection.find({}) を使うとすべてのドキュメントを順に処理できます。また、{"category": "food"} のようなフィルターを渡すと、条件に一致するドキュメントを選択できます。各結果は辞書のように扱えます。item["category"] のように直接参照すると、そのドキュメントにキーがない場合は KeyError が発生します。代替値を使うのが適切な場合は item.get("category", "(not set)") を使ってください。MongoDB の find() はカーソルを返すので、単一のドキュメントとして扱わず、反復処理してください。

このローカルデモを実行すると、フィールドの異なるドキュメントがどのように表示されるかを確認できます。これはオフラインで動作します。前述のとおり、データベースを使う処理は --database と MONGODB_URI を指定して別途有効にします。

このセクションのコードと表示されるラベルや出力は英語のままです。

import os
import sys
import uuid
from datetime import datetime, timezone


def show_documents(documents):
    for item in documents:
        print(item["item_name"], item.get("category", "(not set)"))


def run_database_workflow():
    from pymongo import MongoClient

    uri = os.environ.get("MONGODB_URI")
    if not uri:
        print("Set MONGODB_URI before running with --database.")
        return

    with MongoClient(uri) as client:
        collection = client["user_shopping_list"]["items"]
        many = [
            {"_id": str(uuid.uuid4()), "item_name": "Bread", "category": "food"},
            {"_id": str(uuid.uuid4()), "item_name": "Blender"},
        ]
        many_result = collection.insert_many(many)
        single = {
            "_id": str(uuid.uuid4()),
            "item_name": "Coffee",
            "category": "food",
            "added_at": datetime.now(timezone.utc),
        }
        single_result = collection.insert_one(single)
        inserted_ids = many_result.inserted_ids + [single_result.inserted_id]
        found = collection.find({"_id": {"$in": inserted_ids}})
        show_documents(found)


def main():
    if "--database" in sys.argv:
        run_database_workflow()
    else:
        # This local sample needs no MongoDB server or connection URI.
        show_documents([
            {"item_name": "Bread", "category": "food"},
            {"item_name": "Blender"},
        ])


if __name__ == "__main__":
    main()

Bread food
Blender (not set)

ローカル例の出力は次のとおりです。

データベースを使う処理では、find() が返すカーソルを反復するときにも、同じ安全なアクセス方法を使います。検索結果がない場合、カーソルは行を返さずに終わります。接続の失敗と決めつける前に、フィルターとコレクション名を確認してください。このモジュールをインポートすると関数は定義されますが、main() は呼び出されず、データベース操作も実行されません。