Python から MongoDB を操作するには、MongoDB 公式のドライバである PyMongo を使います。MongoClient で接続し、コレクションに対して insert_one・find・update_one・delete_one を呼ぶのが基本の流れです。この記事では、接続の作り方、追加・検索・更新・削除、インデックスの作成、ページングや集計までを順に説明し、あわせて「接続が遅い」「ObjectId で見つからない」「ユーザー入力でクエリが書き換えられる」といった、実際につまずきやすい点の対処を扱います。
準備：インストールと接続確認
仮想環境を作り、PyMongo をインストールします。MongoDB サーバーは、手元に入れたものでも、クラウドのマネージドサービスでもかまいません。
python -m venv .venv
source .venv/bin/activate # Windows は .venv\Scripts\activate
pip install pymongoこの記事のコードは PyMongo 4 系を前提にしています（2026年10月現在）。3 系向けの古い記事にある insert()・update()・remove()・count() などのメソッドは 4 系で削除されているため、コピーして動かないときは、まずこの点を疑ってください。
import os
from pymongo import MongoClient
from pymongo.errors import ServerSelectionTimeoutError
uri = os.environ.get("MONGODB_URI", "mongodb://127.0.0.1:27017")
client = MongoClient(uri, serverSelectionTimeoutMS=5000)
try:
client.admin.command("ping")
print("connected")
except ServerSelectionTimeoutError as e:
print("接続できません:", e)
db = client["appdb"]
users = db["users"]MongoClient は、作った時点ではまだサーバーに接続しません。最初の操作のときに接続を試み、つながらなければ既定では30秒ほど待ってから例外になります。serverSelectionTimeoutMS を短めに指定し、起動時に ping で確かめておくと、接続先の誤りにすぐ気づけます。
MongoClient は内部に接続プールを持ち、スレッドから共有して使える設計です。アプリ全体で1つ作り、処理のたびに作り直さないようにします。
接続文字列のユーザー名やパスワードはコードに直接書かず、環境変数などから読み込みます。パスワードに @ や : などの記号が含まれる場合は、urllib.parse.quote_plus でエンコードしてから URI に入れる必要があります。また、MongoDB を外部に公開するなら、認証を有効にし、接続元の IP を絞ってください。認証なしで公開されたままのデータベースが狙われる事例は多く報告されています。
追加・検索・更新・削除
MongoDB ではデータを「ドキュメント」と呼び、Python の辞書の形で扱います。データベースやコレクションは、最初に書き込んだ時点で自動的に作られます。
from datetime import datetime, timezone
# 追加
result = users.insert_one({
"email": "taro@example.com",
"name": "Taro",
"tags": ["admin"],
"created_at": datetime.now(timezone.utc),
})
print(result.inserted_id) # 自動で付く _id（ObjectId）
users.insert_many([
{"email": "hanako@example.com", "name": "Hanako", "tags": []},
{"email": "jiro@example.com", "name": "Jiro", "tags": ["editor"]},
])
# 検索
doc = users.find_one({"email": "taro@example.com"})
for d in users.find({"tags": "editor"}, {"_id": 0, "name": 1, "email": 1}):
print(d)
# 更新（$set で指定したフィールドだけを書き換える）
r = users.update_one({"email": "taro@example.com"}, {"$set": {"name": "Taro Yamada"}})
print(r.matched_count, r.modified_count)
# 配列に要素を追加（重複させない）
users.update_one({"email": "jiro@example.com"}, {"$addToSet": {"tags": "reviewer"}})
# 無ければ作り、あれば更新する
users.update_one(
{"email": "saburo@example.com"},
{"$set": {"name": "Saburo"}, "$setOnInsert": {"created_at": datetime.now(timezone.utc)}},
upsert=True,
)
# 削除
users.delete_one({"email": "hanako@example.com"})
# 件数
print(users.count_documents({"tags": "admin"}))注意したいのは次の点です。
- 更新では演算子を使う：
update_one(条件, {"name": "x"})のように演算子なしで渡すとエラーになります。ドキュメント全体を置き換えたいときはreplace_oneを使い、一部だけ変えるときは$setを使います。 - find は遅延評価：
find()が返すのはカーソルで、ループで読み進めたときに少しずつ取得します。list(users.find())とすると全件をメモリに載せるため、大きなコレクションでは避けます。 - 日時は UTC で保存する：MongoDB の日時はミリ秒精度の UTC として保存されます。タイムゾーン付きの
datetimeで書き込み、読み出すときに表示用のタイムゾーンに変換すると混乱しません。既定では読み出した値がタイムゾーン情報なしのdatetimeになるため、MongoClient(uri, tz_aware=True)にしておくと UTC 付きで受け取れます。
ObjectId で検索する
自動で付く _id は文字列ではなく ObjectId 型です。URL などから受け取った文字列で検索するときは、変換が必要です。
from bson import ObjectId
from bson.errors import InvalidId
def find_user_by_id(id_str: str):
try:
oid = ObjectId(id_str)
except InvalidId:
return None
return users.find_one({"_id": oid}){"_id": "65f..."} のように文字列のまま検索すると、エラーにはならず、単に0件になります。「確かにあるのに見つからない」場合は、まず型を確認してください。bson は PyMongo に同梱されているパッケージです。PyPI の別パッケージ bson を入れると衝突するため、追加でインストールしないでください。
ユーザー入力をクエリに入れるときの注意
MongoDB のクエリは辞書なので、Web アプリで受け取った JSON をそのまま条件に使うと、{"password": {"$ne": ""}} のような演算子を送り込まれ、意図しない検索をされる恐れがあります（NoSQL インジェクション）。
def find_by_email(email_input):
# 文字列以外（辞書や配列）は受け付けない
if not isinstance(email_input, str):
raise ValueError("email must be a string")
return users.find_one({"email": email_input})条件に使う値は型を確かめ、期待する型（文字列・数値）に限定します。入力の検証には Pydantic などのライブラリを使うと、型の確認をまとめて書けます。なお、パスワードは平文で保存せず、bcrypt や Argon2 などでハッシュ化したものを保存してください。
インデックスを作る
検索条件に使うフィールドにはインデックスを作ります。インデックスが無いと、コレクションの全件を読む検索になり、件数が増えると急に遅くなります。
from pymongo import ASCENDING, DESCENDING
from pymongo.errors import DuplicateKeyError
users.create_index([("email", ASCENDING)], unique=True)
users.create_index([("tags", ASCENDING), ("created_at", DESCENDING)])
try:
users.insert_one({"email": "taro@example.com", "name": "dup"})
except DuplicateKeyError:
print("このメールアドレスは登録済みです")
# インデックスが使われているかを確認する
plan = users.find({"tags": "admin"}).explain()
print(plan["queryPlanner"]["winningPlan"])create_index は同じ指定のインデックスが既にあれば何もしないため、アプリ起動時に呼んでも問題ありません。ただし、データが多い本番環境で新しいインデックスを作ると負荷がかかるため、アクセスの少ない時間に実施し、事前に mongodump などでバックアップを取っておきます。explain() の結果で COLLSCAN（全件走査）になっていれば、インデックスが使われていません。
並び替え・ページング・集計
# 新しい順に20件ずつ表示する
page, per_page = 2, 20
cursor = (
users.find({}, {"_id": 0, "name": 1, "created_at": 1})
.sort("created_at", DESCENDING)
.skip((page - 1) * per_page)
.limit(per_page)
)
# タグごとの人数を集計する
pipeline = [
{"$unwind": "$tags"},
{"$group": {"_id": "$tags", "count": {"$sum": 1}}},
{"$sort": {"count": -1}},
]
for row in users.aggregate(pipeline):
print(row["_id"], row["count"])skip はページが深くなるほど読み飛ばす件数が増えて遅くなります。件数が多い一覧では、「前のページの最後の created_at より古いもの」を条件にするやり方（キーセット方式）に切り替えると、速度が安定します。
非同期で使う場合
asyncio のアプリから使う場合、これまでは Motor という別ライブラリが使われてきましたが、PyMongo 本体に非同期 API（AsyncMongoClient）が加わり、Motor からの移行が案内されています（2026年10月現在）。移行のスケジュールや API の差は変わる可能性があるため、新しく書くときは PyMongo の公式ドキュメントで現在の推奨を確認してください。
まとめ
MongoClientはアプリ全体で1つ作り、serverSelectionTimeoutMSとpingで接続を早めに確認する- PyMongo 4 系では
insert_one・find・update_one・delete_one・count_documentsを使う - 更新は
$setなどの演算子で行い、_idの検索はObjectIdに変換してから行う - ユーザー入力は型を確かめてからクエリに入れ、演算子を送り込まれないようにする
- 検索に使うフィールドにはインデックスを作り、
explain()で使われているかを確かめる - 認証を有効にし、接続元を絞り、作業前にはバックアップを取る