MongoDB Atlas + PyMongo Patterns

PyMongo 4.8 driver patterns specialized for MongoDB Atlas: connection pool sizing for Atlas tiers, Atlas Search index design for fuzzy matching, aggregation pipelines, and the right way to call sync PyMongo from async FastAPI.

When to use

• Designing a catalog or product-matching service on MongoDB Atlas
• Tuning Atlas Search indexes for full-text or fuzzy product/SKU lookup
• Writing aggregation pipelines with $search, $searchMeta, or $facet
• Bridging sync PyMongo into async FastAPI endpoints (asyncio.to_thread)
• Evaluating a future migration to AsyncMongoClient (Motor is deprecated 2026-05-14)

For modern Python idioms and async fundamentals, see arch-python-modern. For wrapping sync calls in async endpoints, see fastapi-patterns.

Core patterns

Quick reference

# Singleton client (module-level)
mongo_client: MongoClient | None = None

def create_mongo_client():
    global mongo_client
    mongo_client = MongoClient(
        os.environ["MONGODB_URI"],         # mongodb+srv://...
        maxPoolSize=50,
        minPoolSize=5,                     # keep persistent connections warm
        maxIdleTimeMS=30_000,
        serverSelectionTimeoutMS=30_000,   # ≥30s for Atlas failover
        socketTimeoutMS=30_000,
        waitQueueTimeoutMS=5_000,
        retryWrites=True,                  # Atlas defaults
        retryReads=True,
    )

# Atlas Search aggregation (sync PyMongo in async endpoint)
async def search_products(query: str, limit: int = 20) -> list[dict]:
    pipeline = [
        {"$search": {                       # MUST be first stage
            "index": "atlas-search-index",
            "compound": {
                "must": [{"text": {
                    "query": query,
                    "path": ["product_name", "description"],
                    "fuzzy": {"maxEdits": 1, "prefixLength": 2}
                }}],
                "filter": [{"equals": {"path": "status", "value": "active"}}],
            }
        }},
        {"$limit": limit},                  # bound work early
        {"$addFields": {"score": {"$meta": "searchScore"}}},
        {"$project": {"_id": 1, "product_name": 1, "sku": 1, "score": 1}},
    ]
    coll = get_database()["products"]
    return await asyncio.to_thread(lambda: list(coll.aggregate(pipeline)))

Anti-patterns (top 5)

1. MongoClient(uri) per request → singleton; create once in lifespan
2. Unbounded .find()/.aggregate() → always $limit early in the pipeline
3. Missing index on filter field → create_index([("field", ASCENDING)]) or Atlas UI; verify with .explain("executionStats")
4. Sync PyMongo in async def without thread wrap → await asyncio.to_thread(coll.find_one, q)
5. $search not first stage → MongoServerError; pre-filters go inside compound.filter, not as preceding $match

Important: Motor deprecated 2026-05-14

If any code in this project (or sibling services) uses Motor (motor.motor_asyncio.AsyncIOMotorClient), plan migration to AsyncMongoClient (PyMongo 4.9+) before May 14, 2026. The current example-service uses sync PyMongo — no Motor migration needed, but evaluate AsyncMongoClient at the next major version bump.

See reference.md for full pattern catalogue, examples.md for project-grounded implementations.