杰瑞科技汇

Python MongoClient如何连接使用?

Of course! Let's dive into using a Python MongoDB client. The most popular and recommended library for this is pymongo.

Python MongoClient如何连接使用?-图1
(图片来源网络,侵删)

This guide will cover:

  1. Installation: How to get the library.
  2. Connecting: How to connect to a MongoDB instance (local or cloud).
  3. Core Operations: The essential CRUD (Create, Read, Update, Delete) operations.
  4. Best Practices: Using connection strings and environment variables.
  5. Alternative: A brief look at the modern, async alternative motor.

Installation

First, you need to install the pymongo library. It's highly recommended to also install dnspython, which is needed for connecting to MongoDB Atlas (the cloud service) or other hosts that require DNS SRV records.

pip install pymongo "dnspython>=2.0.0"

Connecting to MongoDB

You can connect to a local MongoDB server or a remote one (like MongoDB Atlas).

A. Connecting to a Local Server

If you have MongoDB installed locally on your machine (the default port is 27017), the connection is straightforward.

Python MongoClient如何连接使用?-图2
(图片来源网络,侵删)
from pymongo import MongoClient
# By default, MongoClient connects to localhost:27017
client = MongoClient()
# You can also be explicit:
# client = MongoClient('mongodb://localhost:27017/')
# Check if the connection was successful
try:
    # The ismaster command is cheap and does not require auth.
    client.admin.command('ismaster')
    print("MongoDB connection successful!")
except Exception as e:
    print(f"MongoDB connection failed: {e}")
# IMPORTANT: Always close the connection when you're done
client.close()

B. Connecting to MongoDB Atlas (Cloud)

When using a cloud service like MongoDB Atlas, you get a connection string. Never hardcode this string directly in your code. We'll use environment variables for best practice.

Get Your Connection String: From your Atlas cluster, go to "Database Access" and create a user. Then go to "Network Access" and add your IP address (or 0.0.0/0 for any IP, for development only). Finally, click "Connect" on your cluster and choose "Connect your application". Select Python and copy the connection string.

It will look something like this: mongodb+srv://<username>:<password>@cluster0.xxxxx.mongodb.net/?retryWrites=true&w=majority

Use Environment Variables: Store your username, password, and connection string in a .env file in your project's root directory.

Python MongoClient如何连接使用?-图3
(图片来源网络,侵删)

.env file:

MONGO_URI="mongodb+srv://<username>:<password>@cluster0.xxxxx.mongodb.net/?retryWrites=true&w=majority"

Connect in Python: Use the python-dotenv library to load your environment variables.

First, install it: pip install python-dotenv

Then, in your Python script:

import os
from dotenv import load_dotenv
from pymongo import MongoClient
# Load environment variables from .env file
load_dotenv()
# Get the connection string from the environment variable
mongo_uri = os.getenv("MONGO_URI")
if not mongo_uri:
    raise ValueError("MONGO_URI environment variable not set")
# Connect to MongoDB using the connection string
client = MongoClient(mongo_uri)
# Now you can access databases and collections
db = client['mydatabase'] # Or client.get_database('mydatabase')
# Check connection
try:
    client.admin.command('ping')
    print("Pinged your deployment. You successfully connected to MongoDB!")
except Exception as e:
    print(e)
client.close()

Core CRUD Operations

Let's assume you have a database named mydatabase and a collection named users.

Create (Insert)

from pymongo import MongoClient
# --- Assume 'client' is already connected as shown above ---
db = client['mydatabase']
users_collection = db['users']
# Insert a single document
user1 = {
    "name": "Alice",
    "age": 30,
    "city": "New York",
    "hobbies": ["reading", "hiking"]
}
insert_result = users_collection.insert_one(user1)
print(f"Inserted document with id: {insert_result.inserted_id}")
# Insert multiple documents
users_to_insert = [
    {"name": "Bob", "age": 25, "city": "San Francisco"},
    {"name": "Charlie", "age": 35, "city": "London"}
]
insert_many_result = users_collection.insert_many(users_to_insert)
print(f"Inserted document ids: {insert_many_result.inserted_ids}")

Read (Find)

# --- Assume 'users_collection' is already defined ---
# Find a single document
# The query is a dictionary of field:value pairs
alice = users_collection.find_one({"name": "Alice"})
print(f"Found Alice: {alice}")
# Find all documents matching a query
users_over_30 = users_collection.find({"age": {"$gt": 30}}) # $gt means "greater than"
print("\nUsers over 30:")
for user in users_over_30:
    print(user)
# Find with multiple conditions (AND)
nyc_users = users_collection.find({"city": "New York", "age": {"$lt": 40}}) # $lt means "less than"
print("\nNYC users under 40:")
for user in nyc_users:
    print(user)

Update

# --- Assume 'users_collection' is already defined ---
# Update a single document
# The first argument is the filter, the second is the update operation
update_result = users_collection.update_one(
    {"name": "Bob"},
    {"$set": {"city": "Chicago", "age": 26}} # $set updates fields or adds new ones
)
print(f"Matched {update_result.matched_count} document and modified {update_result.modified_count} document.")
# Update multiple documents
update_many_result = users_collection.update_many(
    {"city": "London"},
    {"$set": {"country": "UK"}}
)
print(f"Matched {update_many_result.matched_count} documents and modified {update_many_result.modified_count} documents.")

Delete

# --- Assume 'users_collection' is already defined ---
# Delete a single document
delete_result = users_collection.delete_one({"name": "Charlie"})
print(f"Deleted {delete_result.deleted_count} document.")
# Delete multiple documents
# Be very careful with delete_many!
delete_many_result = users_collection.delete_many({"age": {"$lt": 30}})
print(f"Deleted {delete_many_result.deleted_count} documents.")

Best Practices & Pro Tips

A. Using a Connection String Directly (Simpler for Quick Scripts)

For simple scripts or learning, you can pass the connection string directly to MongoClient(). Remember not to commit this to a public repository.

from pymongo import MongoClient
# Directly from your Atlas dashboard
CONNECTION_STRING = "mongodb+srv://<user>:<password>@cluster0.xxxxx.mongodb.net/mydatabase?retryWrites=true&w=majority"
# The database name can be part of the URI or specified later
client = MongoClient(CONNECTION_STRING)
db = client.get_database() # Gets 'mydatabase' from the URI

B. Indexing

For better query performance, especially on large collections, create indexes on fields you frequently query.

# Create an index on the 'name' field
users_collection.create_index([("name", 1)]) # 1 for ascending, -1 for descending
# To see all indexes on a collection
print(users_collection.list_indexes())

C. The with Statement for Connection Management

For more robust connection handling, you can use MongoClient as a context manager. It ensures the connection is closed automatically.

from pymongo import MongoClient
MONGO_URI = "mongodb://localhost:27017/"
with MongoClient(MONGO_URI) as client:
    db = client['mydatabase']
    users_collection = db['users']
    # Perform your operations here
    print("Doing some work...")
    # The connection is automatically closed when the 'with' block exits
print("Connection closed.")

The Modern Alternative: motor

If you are building a modern, asynchronous application using asyncio, pymongo's synchronous calls can become a bottleneck. The solution is motor, the async version of pymongo.

It has the same API but uses async and await.

Installation: pip install motor

Example (using asyncio):

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient
async def main():
    # Connect to MongoDB
    client = AsyncIOMotorClient("mongodb://localhost:27017/")
    db = client['mydatabase']
    users_collection = db['users']
    # Insert a document
    await users_collection.insert_one({"name": "AsyncUser", "status": "active"})
    print("Inserted async user.")
    # Find a document
    user = await users_collection.find_one({"name": "AsyncUser"})
    print(f"Found async user: {user}")
    # Close the connection
    client.close()
# Run the async main function
asyncio.run(main())

Summary

Task pymongo (Sync) motor (Async)
Connect client = MongoClient(uri) client = AsyncIOMotorClient(uri)
Get DB db = client['dbname'] db = client['dbname']
Get Collection coll = db['collname'] coll = db['collname']
Insert One coll.insert_one(doc) await coll.insert_one(doc)
Find One doc = coll.find_one(filter) doc = await coll.find_one(filter)
Find Many cursor = coll.find(filter) cursor = await coll.find(filter) (async iterable)
Update One coll.update_one(filter, update) await coll.update_one(filter, update)
Close client.close() client.close() (or use async with)

For most general-purpose Python scripts, pymongo is the perfect choice. For high-performance web servers like FastAPI, Starlette, or aiohttp, motor is the recommended path.

分享:
扫描分享到社交APP
上一篇
下一篇