Flask MongoDB

Introduction

MongoDB is a popular NoSQL database that stores data in flexible, JSON-like documents, making it a great choice for Python and Flask applications. Unlike traditional relational databases like SQLite or PostgreSQL, MongoDB is schema-less, meaning you don't need to define your data structure beforehand, allowing for more flexible development.

In this tutorial, we'll learn how to integrate MongoDB with Flask applications. We'll cover installation, basic CRUD operations (Create, Read, Update, Delete), and build a simple but practical application to demonstrate MongoDB's capabilities.

Prerequisites

Before we start, make sure you have:

Python 3.6+ installed
Basic knowledge of Flask
Basic understanding of databases

Setting Up MongoDB with Flask

Installing Required Packages

First, let's install the necessary packages:

bash
pip install flask pymongo

You'll also need MongoDB installed on your system. You can download it from the MongoDB website or use a cloud service like MongoDB Atlas for a free hosted database.

Connecting to MongoDB

Let's create a basic Flask application that connects to MongoDB:

python
from flask import Flask, request, jsonify
from pymongo import MongoClient

app = Flask(__name__)

# Setup MongoDB connection
client = MongoClient('mongodb://localhost:27017/')
db = client['flask_demo']  # Select the database
collection = db['users']   # Select the collection

@app.route('/')
def home():
    return "Flask MongoDB Integration"

if __name__ == '__main__':
    app.run(debug=True)

This establishes a connection to a MongoDB database named flask_demo with a collection called users. If you're using MongoDB Atlas, your connection string will be different and look something like:

python
client = MongoClient('mongodb+srv://username:[email protected]/myFirstDatabase')

Basic CRUD Operations with Flask and MongoDB

Let's implement the four basic operations that most applications need:

1. Create (Insert) Data

python
@app.route('/users/add', methods=['POST'])
def add_user():
    data = request.get_json()
    
    if not data:
        return jsonify({"error": "Invalid data"}), 400
        
    # Insert the user data into MongoDB
    user_id = collection.insert_one(data).inserted_id
    
    # Convert ObjectId to string for JSON response
    return jsonify({"message": "User added successfully", 
                    "user_id": str(user_id)}), 201

2. Read (Query) Data

python
@app.route('/users', methods=['GET'])
def get_users():
    users = list(collection.find({}, {'_id': 0}))  # Exclude MongoDB _id field
    return jsonify(users), 200

@app.route('/users/<username>', methods=['GET'])
def get_user(username):
    user = collection.find_one({'username': username}, {'_id': 0})
    
    if user:
        return jsonify(user), 200
    return jsonify({"error": "User not found"}), 404

3. Update Data

python
@app.route('/users/<username>', methods=['PUT'])
def update_user(username):
    data = request.get_json()
    
    if not data:
        return jsonify({"error": "Invalid data"}), 400
        
    result = collection.update_one(
        {'username': username}, 
        {'$set': data}
    )
    
    if result.matched_count:
        return jsonify({"message": "User updated successfully"}), 200
    return jsonify({"error": "User not found"}), 404

4. Delete Data

python
@app.route('/users/<username>', methods=['DELETE'])
def delete_user(username):
    result = collection.delete_one({'username': username})
    
    if result.deleted_count:
        return jsonify({"message": "User deleted successfully"}), 200
    return jsonify({"error": "User not found"}), 404

Advanced MongoDB Features

MongoDB offers several advanced features that make it powerful for web applications:

1. Document Embedding

One of MongoDB's strengths is the ability to embed related data in a single document:

python
# Example of a user document with embedded address information
user_with_addresses = {
    "username": "johndoe",
    "email": "[email protected]",
    "addresses": [
        {
            "type": "home",
            "street": "123 Main St",
            "city": "New York",
            "zip": "10001"
        },
        {
            "type": "work",
            "street": "456 Market St",
            "city": "New York",
            "zip": "10003"
        }
    ]
}

# Insert the user with embedded addresses
collection.insert_one(user_with_addresses)

2. Querying Embedded Documents

You can easily query embedded documents using dot notation:

python
# Find users with a home address in New York
users_in_ny = collection.find({
    "addresses.type": "home",
    "addresses.city": "New York"
})

3. Aggregation Framework

MongoDB's aggregation framework lets you process data and get computed results:

python
@app.route('/stats/city', methods=['GET'])
def get_city_stats():
    pipeline = [
        {"$unwind": "$addresses"},
        {"$group": {
            "_id": "$addresses.city",
            "count": {"$sum": 1},
            "users": {"$addToSet": "$username"}
        }},
        {"$sort": {"count": -1}}
    ]
    
    results = list(collection.aggregate(pipeline))
    return jsonify(results)

Building a Complete User Management API

Let's put everything together into a complete example of a user management API:

python
from flask import Flask, request, jsonify
from pymongo import MongoClient
from bson.objectid import ObjectId
import datetime

app = Flask(__name__)

# Setup MongoDB connection
client = MongoClient('mongodb://localhost:27017/')
db = client['user_management']
users = db['users']

# Ensure we have an index on username for faster lookups
users.create_index("username", unique=True)

@app.route('/')
def home():
    return "User Management API with Flask and MongoDB"

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    
    # Basic validation
    if not data or not all(key in data for key in ['username', 'email']):
        return jsonify({"error": "Missing required fields"}), 400
    
    # Check if user already exists
    if users.find_one({"username": data['username']}):
        return jsonify({"error": "Username already exists"}), 409
    
    # Add creation timestamp
    data['created_at'] = datetime.datetime.utcnow()
    
    try:
        result = users.insert_one(data)
        return jsonify({"message": "User created successfully", 
                        "user_id": str(result.inserted_id)}), 201
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/users', methods=['GET'])
def get_all_users():
    limit = int(request.args.get('limit', 10))
    skip = int(request.args.get('skip', 0))
    
    user_list = list(users.find({}, {'_id': 0}).limit(limit).skip(skip))
    return jsonify({"users": user_list, "count": len(user_list)}), 200

@app.route('/users/<username>', methods=['GET'])
def get_user(username):
    user = users.find_one({"username": username}, {"_id": 0})
    
    if user:
        return jsonify(user), 200
    return jsonify({"error": "User not found"}), 404

@app.route('/users/<username>', methods=['PUT'])
def update_user(username):
    data = request.get_json()
    
    if not data:
        return jsonify({"error": "No data provided"}), 400
    
    # Don't allow username updates through this endpoint
    if 'username' in data:
        del data['username']
    
    # Add update timestamp
    data['updated_at'] = datetime.datetime.utcnow()
    
    result = users.update_one(
        {"username": username},
        {"$set": data}
    )
    
    if result.matched_count:
        return jsonify({"message": "User updated successfully"}), 200
    return jsonify({"error": "User not found"}), 404

@app.route('/users/<username>', methods=['DELETE'])
def delete_user(username):
    result = users.delete_one({"username": username})
    
    if result.deleted_count:
        return jsonify({"message": "User deleted successfully"}), 200
    return jsonify({"error": "User not found"}), 404

@app.route('/users/search', methods=['GET'])
def search_users():
    query = request.args.get('q', '')
    
    if not query:
        return jsonify({"error": "Search query required"}), 400
    
    # Search in multiple fields
    search_results = list(users.find({
        "$or": [
            {"username": {"$regex": query, "$options": "i"}},
            {"email": {"$regex": query, "$options": "i"}},
            {"name": {"$regex": query, "$options": "i"}}
        ]
    }, {"_id": 0}))
    
    return jsonify({"results": search_results, "count": len(search_results)}), 200

if __name__ == '__main__':
    app.run(debug=True)

This API provides endpoints for creating, reading, updating, and deleting users, along with search capability.

Testing Your MongoDB Flask Integration

You can test your API using tools like Postman, curl, or even write simple Python scripts.

Here's an example using curl:

bash
# Create a new user
curl -X POST -H "Content-Type: application/json" -d '{"username": "john_doe", "email": "[email protected]", "name": "John Doe"}' http://localhost:5000/users

# Get all users
curl http://localhost:5000/users

# Get a specific user
curl http://localhost:5000/users/john_doe

# Update a user
curl -X PUT -H "Content-Type: application/json" -d '{"email": "[email protected]"}' http://localhost:5000/users/john_doe

# Delete a user
curl -X DELETE http://localhost:5000/users/john_doe

# Search for users
curl http://localhost:5000/users/search?q=john

Best Practices for Flask-MongoDB Applications

When working with Flask and MongoDB, keep these best practices in mind:

Connection Management: Use connection pooling and close connections when done.
Error Handling: Always handle potential MongoDB exceptions.
Indexing: Create indexes for fields you'll query frequently.
Data Validation: Validate data before inserting or updating.
Security: Never expose direct MongoDB queries to user input.
Environment Variables: Store your MongoDB connection string in environment variables.

Here's how to implement some of these best practices:

python
import os
from flask import Flask
from pymongo import MongoClient
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

app = Flask(__name__)

# Use environment variables for MongoDB connection
MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017/")
DB_NAME = os.getenv("DB_NAME", "flask_app")

# Create a MongoDB client as a global variable
mongo_client = MongoClient(MONGO_URI)
db = mongo_client[DB_NAME]

# Register a teardown function to close the connection
@app.teardown_appcontext
def close_db_connection(error):
    mongo_client.close()

Summary

In this tutorial, we've covered:

How to set up MongoDB with Flask
Performing basic CRUD operations with PyMongo
Advanced MongoDB features like document embedding and aggregation
Building a complete user management API
Best practices for Flask-MongoDB applications

MongoDB offers a flexible, scalable solution for Flask applications that need to store complex, evolving data structures. Its schema-less design is particularly useful during rapid development phases when your data model might change frequently.

Additional Resources

Exercises

Exercise 1: Extend the user management API to include user authentication with password hashing.
Exercise 2: Add pagination to the GET /users endpoint.
Exercise 3: Create a new endpoint that allows filtering users by multiple criteria.
Exercise 4: Implement a logging system that tracks all changes to user data in a separate collection.
Exercise 5: Build a simple blogging system where users can create, read, update, and delete their blog posts.

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)

Introduction​

Prerequisites​

Setting Up MongoDB with Flask​

Installing Required Packages​

Connecting to MongoDB​

Basic CRUD Operations with Flask and MongoDB​

1. Create (Insert) Data​

2. Read (Query) Data​

3. Update Data​

4. Delete Data​

Advanced MongoDB Features​

1. Document Embedding​

2. Querying Embedded Documents​

3. Aggregation Framework​

Building a Complete User Management API​

Testing Your MongoDB Flask Integration​

Best Practices for Flask-MongoDB Applications​

Summary​

Additional Resources​

Exercises​