اگر داده‌های پروژه‌تان ساختار ثابتی ندارند، MongoDB انتخاب رایجی است؛ و اگر با پایتون کار می‌کنید، PyMongo درایور رسمی و مستند MongoDB برای پایتون است. در این راهنما از صفر نصب را با pip یاد می‌گیرید، اولین اتصال را می‌سازید، سند می‌نویسید و می‌خوانید، ایندکس می‌زنید و در نهایت نسخه ناهمگام یعنی AsyncMongoClient را برای پروژه‌های asyncio و FastAPI به کار می‌برید. نسخه فعلی PyMongo، هم API همگام و هم API ناهمگام را داخل یک پکیج ارائه می‌دهد و دیگر لازم نیست برای کارهای ناهمگام حتماً سراغ Motor بروید.

PyMongo چیست و برای کار با MongoDB چه کاری انجام می‌دهد؟

PyMongo درایور رسمی پایتون برای MongoDB است: با آن به سرور وصل می‌شوید، کالکشن می‌سازید، سند (دیکشنری) درج و به‌روزرسانی می‌کنید، کوئری می‌زنید، ایندکس و Aggregation Pipeline تعریف می‌کنید و در نسخه ۴ با کلاس AsyncMongoClient همان کارها را به‌صورت ناهمگام داخل رویدادلوپ اجرا می‌کنید. اگر تا حالا با SQL کار کرده‌اید، ادامه‌ی همین مقاله را بخوانید؛ در پایان یک چک‌لیست عملی برای اتصال امن و مهاجرت از Motor هم می‌گیرید.

فهرست مطالب

MongoDB چیست و چه زمانی از PyMongo استفاده کنیم؟
پیش‌نیازها و نصب PyMongo 4
اولین اتصال و نوشتن اولین سند با MongoClient
خواندن، فیلتر و ایندکس: کوئری‌های روزمره
AsyncMongoClient برای پروژه‌های ناهمگام و FastAPI
مقایسه PyMongo با Motor و درایورهای مشابه
اشتباهات رایج در کار با PyMongo
امنیت و کاربرد واقعی در پروژه‌های هوش مصنوعی
مسیر ادامه یادگیری
سوالات متداول
جمع‌بندی

MongoDB چیست و چه زمانی از PyMongo استفاده کنیم؟

MongoDB یک پایگاه داده سندی (Document Database) است: به‌جای «جدول» و «ردیف»، با «کالکشن» و «سند» کار می‌کنید و هر سند یک دیکشنری JSON مانند است. همین ویژگی باعث می‌شود وقتی اسکیمای داده مدام تغییر می‌کند، مجبور نباشید برای هر فیلد جدید مایگریشن بنویسید. PyMongo هم درایور سمت پایتون این پایگاه داده است و کتابخانه bson و ابزار gridfs را هم همراه خود می‌آورد.

در پایتون معمولاً در سه موقعیت سراغ PyMongo می‌روید: کار با داده‌های نیمه‌ساختاریافته (خروجی APIهای بیرونی، لاگ، رویداد)، ذخیره اسناد بزرگ و تودرتو (پروفایل کاربر، سند سفارش با اقلامش)، و پروژه‌های هوش مصنوعی که می‌خواهند تاریخچه چت، کش پاسخ‌ها یا متادیتای بردارها را نگه دارند. اگر داده شما کاملاً رابطه‌ای است، گزارش‌های پیچیده می‌خواهید یا به تراکنش چندسندی روی یک جدول وابسته‌اید، یک پایگاه داده رابطه‌ای مثل PostgreSQL انتخاب امن‌تری است.

یک نکته مهم که خیلی‌ها را غافلگیر می‌کند: تراکنش‌های چندسندی در MongoDB روی Replica Set کار می‌کنند، نه روی یک سرور تک (standalone). اگر با docker run mongo سرور محلی بالا می‌آورید و می‌خواهید تراکنش تست کنید، باید سرور را به‌صورت تک‌نودی با قابلیت Replica Set بالا بیاورید.

پیش‌نیازها و نصب PyMongo 4

پیش‌نیاز ساده است: یک پایتون به‌روز (PyMongo برای CPython 3.9 به بالا و PyPy 3.9 به بالا ساخته شده) و یک سرور MongoDB در دسترس؛ یا نصب محلی، یا یک کلاستر ابری. نصب درایور با pip انجام می‌شود:

python -m pip install --upgrade pymongo
python -c "import pymongo, sys; print(pymongo.version); print(sys.version)"

خروجی خط دوم، نسخه نصب‌شده درایور را نشان می‌دهد. نسخه فعلی سری ۴ است و از MongoDB 4.4 تا 9.0 و از API همگام و ناهمگام پشتیبانی می‌کند. اگر پروژه‌تان چند نفر روی آن کار می‌کنند، نسخه را در فایل وابستگی‌ها قفل کنید تا همه روی یک نسخه باشند:

# requirements.txt
pymongo>=4.11,<5

برای اجرای یک سرور محلی در چند ثانیه، داکر ساده‌ترین راه است. تگ تصویر را با نسخه‌ای که در سرور شما پشتیبانی می‌شود هماهنگ کنید:

docker run -d --name mongo-dev -p 27017:27017 mongo:8
docker ps

اگر Docker ندارید، نسخه Community را مستقیم نصب کنید و سرویس را روی پورت پیش‌فرض ۲۷۰۱۷ اجرا کنید. نصب موتور یا ابزارهای مدیریتی اضافه مثل Compass لازم نیست؛ برای دیدن داده‌ها همین اسکریپت‌های پایتون کافی است.

اولین اتصال و نوشتن اولین سند با MongoClient

در PyMongo سطح‌ها به ترتیب چنین هستند: کلاینت → دیتابیس → کالکشن. یک شیء MongoClient را بسازید و از آن دیتابیس و کالکشن بگیرید؛ کالکشن اگر وجود نداشته باشد با اولین درج ساخته می‌شود:

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017")
db = client["shop"]
orders = db["orders"]

result = orders.insert_one({
    "customer": "سارا",
    "total": 129000,
    "items": [{"sku": "A-1", "qty": 2}, {"sku": "B-7", "qty": 1}],
    "status": "paid"
})
print(result.inserted_id)

doc = orders.find_one({"customer": "سارا"})
print(doc["total"], doc["items"][0]["sku"])

client.close()

چند نکته که همین‌جا جلوی دردسر بعدی را می‌گیرد: اول اینکه client «آماده» و امن برای استفاده هم‌زمان چند نخ است و استخر اتصال (connection pool) دارد؛ پس آن را یک بار بسازید و نگه دارید، نه در هر درخواست. دوم اینکه هر سند یک _id می‌گیرد که اگر ندهید، درایور خودش یک ObjectId می‌سازد و در همان بازگشتی به شما می‌دهد. سوم اینکه اتصال ساخته شدن و بلافاصله خطا نمی‌دهد؛ اتصال واقعی در پس‌زمینه انجام می‌شود، پس برای تست سلامت می‌توانید ping بزنید:

from pymongo.errors import ConnectionFailure

try:
    client.admin.command("ping")
    print("اتصال برقرار است")
except ConnectionFailure as exc:
    print("سرور در دسترس نیست:", exc)

خواندن، فیلتر و ایندکس: کوئری‌های روزمره

بیشتر کوئری‌های واقعی سه تکه دارند: فیلتر، مرتب‌سازی و محدودکردن تعداد نتیجه. ایندکس را قبل از اجرا بسازید تا از collection scan فرار کنید:

orders.create_index([("customer", 1), ("created_at", -1)])
orders.create_index([("total", -1)], name="total_desc")

cursor = (
    orders
    .find(
        {"status": "paid", "total": {"$gte": 100000}},
        {"_id": 0, "customer": 1, "total": 1}
    )
    .sort("total", -1)
    .limit(5)
)

for order in cursor:
    print(order)

print(orders.count_documents({"customer": "سارا"}))
print(orders.estimated_document_count())

در این کد چند عملگر پرکاربرد را می‌بینید: 1 یعنی صعودی و -1 یعنی نزولی (برای تاریخ‌ها معمولاً نزولی می‌خواهید)، $gte یعنی «بزرگ‌تر یا مساوی»، و دیکشنری دوم پارامتر projection است که فیلدهای غیرضروری را حذف می‌کند تا پهنای باند کم شود. count_documents شمارش دقیق می‌دهد ولی کندتر است؛ estimated_document_count از فراداده‌ها می‌خواند و تقریبی و بسیار سریع است.

برای کوئری‌های تحلیلی از Aggregation Pipeline استفاده کنید و برای کوئری کند، اول explain بگیرید تا ببینید کدام ایندکس استفاده شده یا اصلا اسکن کامل اتفاق می‌افتد:

pipeline = [
    {"$match": {"status": "paid"}},
    {"$group": {"_id": "$customer", "sum": {"$sum": "$total"}}},
    {"$sort": {"sum": -1}}
]

for row in orders.aggregate(pipeline):
    print(row)

plan = orders.find({"customer": "سارا"}).explain()
print(plan["queryPlanner"]["winningPlan"])

AsyncMongoClient برای پروژه‌های ناهمگام و FastAPI

از نسخه ۴ به بعد، خود PyMongo یک درایور ناهمگام کامل دارد و دیگر لازم نیست بسته جداگانه‌ای نصب کنید. کافی است AsyncMongoClient را از همان پکیج pymongo ایمپورت کنید. الگوی درست این است: هر عملیات I/O را await کنید، کورسور را با async for بخوانید و کلاینت را با async with ببندید.

import asyncio
from pymongo import AsyncMongoClient

async def main():
    async with AsyncMongoClient("mongodb://localhost:27017") as client:
        orders = client["shop"]["orders"]

        await orders.insert_one({"customer": "علی", "total": 45000, "status": "paid"})

        async for order in orders.find({"total": {"$gte": 40000}}):
            print(order["customer"], order["total"])

        print("تعداد سفارش‌ها:", await orders.count_documents({}))

asyncio.run(main())

تفاوت‌های کلیدی با نسخه همگام را جدی بگیرید: با فراخوانی await روی متدی که قابل انتظار نیست، یا با for معمولی روی کورسور ناهمگام، کد شما یا خطای TypeError می‌دهد یا بی‌صدا هیچ رکوردی پردازش نمی‌کند. همچنین پارامتر connect در کلاینت ناهمگام پشتیبانی نمی‌شود، چون درایور نباید نخ پس‌زمینه بسازد؛ به‌جای آن از async with استفاده کنید. نشست (session) هم قابل استفاده هم‌زمان یا بین نخ‌ها نیست، پس هر درخواست را با نشست خودش جلو ببرید.

در FastAPI کلاینت را یک بار در ماژول بسازید و به‌عنوان وابستگی استفاده کنید تا هر درخواست کانکشن جدید باز نکند:

from fastapi import Depends, FastAPI
from pymongo import AsyncMongoClient
from pymongo.asynchronous.database import AsyncDatabase

client = AsyncMongoClient("mongodb://localhost:27017")
app = FastAPI()

def get_db() -> AsyncDatabase:
    return client["shop"]

@app.get("/orders")
async def list_orders(db: AsyncDatabase = Depends(get_db)):
    return [doc async for doc in db["orders"].find({"status": "paid"})]

مقایسه PyMongo با Motor و درایورهای مشابه

سؤال «PyMongo بهتر است یا Motor؟» الان پاسخ روشنی دارد. Motor به‌طور رسمی در ۱۴ اردیبهشت ۲۰۲۶ منسوخ (deprecated) اعلام شده و تنها تا ۱۴ اردیبهشت ۲۰۲۷ باگ‌های بحرانی آن اصلاح می‌شود؛ یعنی پروژه‌های جدید باید سراغ درایور ناهمگام داخل خود PyMongo بروند. جدول زیر را نگه دارید تا برای تصمیم‌های معماری و بازنگری کد قدیمی سریع به آن نگاه کنید.

معیار	PyMongo همگام	PyMongo ناهمگام	Motor
نصب	در همان بسته pymongo	در همان بسته pymongo	بسته جداگانه motor
کلاس کلاینت	MongoClient	AsyncMongoClient	AsyncIOMotorClient
الگوی استفاده	فراخوانی مستقیم متد	await روی متد و async for روی کورسور	await روی متد و async for روی کورسور
وضعیت نگهداری	پشتیبانی رسمی	پشتیبانی رسمی	منسوخ؛ فقط باگ‌فیکس بحرانی
مناسب برای	اسکریپت، کران‌جاب، کد همگام	FastAPI، asyncio، بار هم‌زمان بالا	فقط کد قدیمی که هنوز مهاجرت نشده

اشتباهات رایج در کار با PyMongo

ساختن کلاینت تازه در هر تابع یا هر درخواست: هر کلاینت استخر اتصال خودش را می‌سازد و پس از پایان کار، اتصال‌ها را رها می‌کند؛ نتیجه، کندی و خطاهای شبکه است. یک کلاینت در سطح اپ بسازید.
قاطی کردن الگوی همگام و ناهمگام: با for معمولی روی خروجی find ناهمگام، هیچ رکوردی پردازش نمی‌شود؛ از async for استفاده کنید.
کوئری بدون ایندکس: فیلتر روی فیلدی که ایندکس ندارد یعنی اسکن کل کالکشن. با explain بررسی کنید و ایندکس بسازید.
شمارش گران روی کالکشن بزرگ: count_documents روی میلیون‌ها سند واقعاً اسکن می‌کند؛ برای نمایش تعداد کل از estimated_document_count استفاده کنید.
دور زدن لایه درایور: کار با raw_bson یا ساخت دستی کوئری، به‌جای بهینه‌سازی واقعی، کد را غیرقابل‌نگهداری می‌کند؛ اول ایندکس و پروجکشن را درست کنید.
نگه‌داشتن رمزها در کد: رشته اتصال را داخل فایل کد یا مخزن ننویسید؛ از متغیر محیطی استفاده کنید.

امنیت و کاربرد واقعی در پروژه‌های هوش مصنوعی

در محیط تولید، رشته اتصال را از متغیر محیطی بگیرید و اگر رمز عبور کاراکترهای خاص مثل : یا @ دارد، آن‌ها را با urllib.parse.quote_plus کدگذاری کنید. برای سرویس‌های ابری مثل Atlas معمولاً یک رشته mongodb+srv:// می‌گیرید که TLS به‌صورت پیش‌فرض فعال است و برای رمزگذاری آن به بسته dnspython نیاز دارید. نکات عملی: با کاربر کم‌اختیاز (least privilege) وصل شوید، پورت ۲۷۰۱۷ را روی اینترنت باز نکنید، و قابلیت‌های retryReads و retryWrites را که در PyMongo ۴ پیش‌فرض روشن‌اند حفظ کنید تا خطاهای گذرا خودکار بازیابی شوند.

یکی از کاربردهای پرتقاضای MongoDB در پروژه‌های هوش مصنوعی، لایه داده کمکی است: ذخیره تاریخچه چت و پیام‌های سیستمی، کش نتایج پرس‌وجو تا تکرار دوباره یک تمپرچر دیرهنگام هزینه نداشته باشد، نگه‌داشتن متادیتا و وضعیت پردازش اسناد پیش از فرستادن بردارها به پایگاه داده برداری اختصاصی، و ثبت وضعیت و کلید ادامه (checkpoint) برای کارهای دسته‌ای طولانی. دلیلش این است که این داده‌ها نه کاملاً ساختاریافته‌اند و نه فقط یک کلید و یک بردار. اگر همین الگو را در مقیاس بزرگ‌تر می‌خواهید، به‌جای نوشتن لایه ذخیره‌سازی روی فایل از همین نوع کالکشن‌های اسنادی شروع کنید؛ مهاجرت به دیتابیس برداری بعداً ساده‌تر خواهد بود.

مسیر ادامه یادگیری

بعد از تسلط بر CRUD و ایندکس، سه مسیر پرطرفدار دارید: Aggregation Pipeline برای گزارش‌گیری و کار با داده‌های بزرگ، gridfs از بسته PyMongo برای فایل‌های بزرگ‌تر از ۱۶ مگابایت، و Change Stream برای واکنش به تغییرات به‌صورت زنده. برای صحت داده هم می‌توانید خروجی را با Pydantic اعتبارسنجی کنید. برای تمرین روی کتابخانه‌های پردازش تصویر با پایتون هم آموزش OpenCV در پایتون را ببینید؛ همان الگوی نصب در محیط مجازی و اجرای اسکریپت‌های کوتاه را اینجا هم به کار بردیم.

سوالات متداول

آیا PyMongo از نسخه‌های قدیمی MongoDB هم پشتیبانی می‌کند؟

درایور سری ۴ از MongoDB 4.4 تا 9.0 را پشتیبانی می‌کند. برای سرورهای خیلی قدیمی‌تر باید سراغ نسخه‌های قدیمی‌تر PyMongo بروید، ولی برای هر پروژه جدید بهتر است اول سرور را به نسخه پشتیبانی‌شده ارتقا دهید.

تفاوت AsyncMongoClient و Motor در چیست؟

Motor یک درایور جداگانه بود که API ناهمگام روی asyncio و Tornado داشت و از ۱۴ اردیبهشت ۲۰۲۶ منسوخ اعلام شده است. AsyncMongoClient همان قابلیت ناهمگام را داخل خود PyMongo می‌دهد و راه پیشنهادی برای پروژه‌های جدید است؛ مهاجرت معمولاً فقط تغییر نام کلاس و ایمپورت است.

برای شروع، Atlas لازم است یا MongoDB محلی کافی است؟

MongoDB محلی برای یادگیری و توسعه کاملاً کافی است و با داکر در چند ثانیه بالا می‌آید. Atlas را زمانی انتخاب کنید که به کلاستور واقعی، بکاپ و لایه رایگان نیاز دارید؛ رشته اتصال آن را از متغیر محیطی بگیرید و داخل کد نگذارید.

چرا کوئری من کند است؟

سه دلیل رایج دارد: نبود ایندکس روی فیلد فیلتر، برگرداندن همه فیلدها بدون پروجکشن، و کوئری‌هایی که با $or یا بدون ایندکس مناسب مجبور به اسکن کل کالکشن می‌شوند. با explain نوع اسکن و ایندکس انتخاب‌شده را ببینید و بعد پروجکشن و ایندکس را اضافه کنید.

برای استفاده از تراکنش چه چیزی لازم است؟

تراکنش‌های چندسندی روی سرور standalone کار نمی‌کنند؛ به یک Replica Set نیاز دارید. در توسعه می‌توانید یک نود را با قابلیت Replica Set بالا بیاورید و در تولید از کلاستور واقعی استفاده کنید.

برای فایل و تصویر داخل MongoDB چه کنم؟

برای فایل‌های کوچک می‌توانید فیلدهای BSON مانند Binary را ذخیره کنید، اما برای فایل‌های بزرگ بهتر است از gridfs که همراه PyMongo استفاده کنید تا فایل در قطعات ذخیره شود و محدودیت اندازه سند مانع کارتان نشود.

جمع‌بندی

درایور را با pip install pymongo نصب کنید و نسخه را در فایل وابستگی‌ها قفل کنید.
برای کد همگام از MongoClient و برای کد ناهمگام از AsyncMongoClient استفاده کنید؛ نیازی به Motor نیست.
یک کلاینت در سطح اپ بسازید و با async with یا close() آزادش کنید.
قبل از کوئری‌های سنگین ایندکس بسازید و با پروجکشن پهنای باند را کم کنید.
با explain سرعت کوئری را بسنجید و برای گزارش‌گیری از Aggregation Pipeline کمک بگیرید.
رشته اتصال را از متغیر محیطی بگیرید و با کاربر کم‌اختیاز و TLS وصل شوید.

با همین شش نکته یک اتصال امن و سریع دارید که از اسکریپت‌های کوچک تا سرویس‌های پرترافیک و ناهمگام قابل رشد است؛ بقیه مسیر را با gridfs، Change Stream و Aggregation Pipeline ادامه دهید.

آموزش MongoDB در پایتون با PyMongo 4؛ AsyncMongoClient