43.1 Introduction to NoSQL
NoSQL means a group of database systems that do not depend only on traditional relational tables. NoSQL databases may store documents, key-value pairs, graphs, or wide-column records. They are useful when application data changes frequently, contains nested structures, needs very fast access, or must be distributed across several database servers.
Example: Flexible NoSQL-style documents
# One document can contain basic customer information.
customer_one = {
"name": "Sara",
"email": "sara@example.com"
}
# Another document can contain additional fields.
customer_two = {
"name": "Michael",
"email": "michael@example.com",
"grade": 7,
"subjects": ["Python", "Databases"]
}
# Display both flexible documents.
print(customer_one)
print(customer_two)
Output:
{'name': 'Sara', 'email': 'sara@example.com'}
{'name': 'Michael', 'email': 'michael@example.com', 'grade': 7, 'subjects': ['Python', 'Databases']}
Output Explanation:
The two documents do not have exactly the same fields. A document database can allow this kind of flexible structure, although applications should still define clear and consistent data rules.
43.2 SQL vs NoSQL
SQL databases usually organize information into related tables with predefined columns. NoSQL databases may use documents or key-value structures that allow more flexible shapes. SQL databases are often selected for strong relationships and structured transactions. NoSQL databases are often selected for flexible records, high-speed access, distributed workloads, or specialized storage requirements.
Example: SQL-style customer information
customers table
+----+---------+---------------------+
| id | name | email |
+----+---------+---------------------+
| 1 | Sara | sara@example.com |
| 2 | Michael | michael@example.com |
+----+---------+---------------------+
Example: NoSQL-style customer information
{
"_id": 1,
"name": "Sara",
"email": "sara@example.com"
}
{
"_id": 2,
"name": "Michael",
"email": "michael@example.com",
"subjects": [
"Python",
"Databases"
]
}
Comparison Output:
SQL:
- Tables
- Rows
- Columns
- Foreign keys
- Joins
NoSQL:
- Documents or specialized structures
- Flexible fields
- Embedded information
- Application-specific data design
Output Explanation:
Neither database style is automatically better for every project. The correct choice depends on data relationships, query requirements, consistency rules, expected growth, and application design.
43.3 Document Databases
A document database stores records as document-like structures containing field names and values. Documents can contain strings, numbers, Boolean values, lists, nested documents, dates, and identifiers. Related information can sometimes be embedded inside one document. Documents are grouped into collections, which are similar to groups of related records.
Example: Product document
# Create a nested product document.
product = {
"name": "Laptop",
"price": 899.99,
"quantity": 8,
"available": True,
"specifications": {
"memory": "16 GB",
"storage": "512 GB",
"screen": "15 inch"
},
"categories": [
"Computers",
"Electronics"
]
}
# Read nested information.
print(product["name"])
print(product["specifications"]["memory"])
print(product["categories"][0])
Output:
Laptop
16 GB
Computers
Output Explanation:
The product contains simple fields, a nested specifications document, and a categories list. Document databases can store this related information together.
43.4 MongoDB Introduction
MongoDB is a document database that organizes information into databases, collections, and documents. A database contains collections, and a collection contains documents. Each document normally receives a unique _id value. MongoDB documents use a BSON representation that supports document fields, arrays, dates, numbers, object identifiers, and nested information.
Example: MongoDB organization
Database: store
Collections:
- customers
- products
- orders
Example products document:
{
"_id": ObjectId("..."),
"name": "Keyboard",
"price": 49.99,
"quantity": 10
}
Structure:
MongoDB server
└── store database
├── customers collection
├── products collection
└── orders collection
Output Explanation:
The store database separates different kinds of information into collections. Each collection contains documents representing individual records.
43.5 Connecting Python to MongoDB
Python applications commonly connect to MongoDB with the official PyMongo driver. A MongoClient represents the connection to the MongoDB deployment. The application then selects a database and collection. Connection information should normally be loaded from environment variables, and the program can use a ping command to confirm that the server is reachable.
Example: Install PyMongo
python -m pip install pymongo
Example: Connect to a local MongoDB server
# Import the MongoDB client.
from pymongo import MongoClient
# Create a connection to the local server.
client = MongoClient(
"mongodb://localhost:27017/",
serverSelectionTimeoutMS=5000
)
try:
# Send a simple command to verify the connection.
client.admin.command("ping")
print("Connected to MongoDB.")
# Select a database.
database = client["school"]
# Select a collection.
students = database["students"]
print("Database:", database.name)
print("Collection:", students.name)
finally:
# Close the client.
client.close()
Output:
Connected to MongoDB.
Database: school
Collection: students
Output Explanation:
The ping confirms that MongoDB is reachable. Selecting a database or collection object does not necessarily create stored data until a write operation occurs.
43.6 Creating Documents
MongoDB documents are commonly created from Python dictionaries. The insert_one() method inserts one document, while insert_many() inserts several documents. When a document does not already contain an _id, MongoDB generates one. The returned result object provides the inserted identifier or identifiers.
Example: Insert one document
# Import the MongoDB client.
from pymongo import MongoClient
client = MongoClient(
"mongodb://localhost:27017/"
)
database = client["store"]
products = database["products"]
# Create a product document.
product = {
"name": "Keyboard",
"price": 49.99,
"quantity": 10,
"available": True
}
# Insert the document.
result = products.insert_one(product)
# Display the generated identifier.
print("Inserted ID:", result.inserted_id)
client.close()
Example Output:
Inserted ID: 669b6f2f8d42d72480f2a101
Output Explanation:
MongoDB creates an ObjectId for the document. The exact identifier is different every time a new document is inserted.
Example: Insert several documents
# Create several product documents.
new_products = [
{
"name": "Mouse",
"price": 24.50,
"quantity": 25
},
{
"name": "Monitor",
"price": 189.00,
"quantity": 8
}
]
# Insert all documents.
result = products.insert_many(new_products)
print("Documents inserted:", len(result.inserted_ids))
Output:
Documents inserted: 2
Output Explanation:
Two dictionaries are inserted as separate MongoDB documents. The result contains one generated identifier for each document.
43.7 Reading Documents
MongoDB documents are read with methods such as find_one() and find(). The first returns one matching document or None. The second returns a cursor that can be looped over. A projection can limit which fields are returned, reducing unnecessary information and making query results easier to work with.
Example: Read one document
# Find one product by name.
product = products.find_one(
{"name": "Keyboard"}
)
if product:
print(product["name"])
print(product["price"])
print(product["quantity"])
else:
print("Product not found.")
Output:
Keyboard
49.99
10
Output Explanation:
The query filter asks MongoDB to find a document whose name equals Keyboard. The returned document behaves like a Python dictionary.
Example: Read several documents
# Return selected fields from all products.
cursor = products.find(
{},
{
"_id": 0,
"name": 1,
"price": 1
}
)
for product in cursor:
print(product)
Example Output:
{'name': 'Keyboard', 'price': 49.99}
{'name': 'Mouse', 'price': 24.5}
{'name': 'Monitor', 'price': 189.0}
Output Explanation:
The empty filter selects all documents. The projection includes name and price while excluding the automatically stored _id field.
43.8 Updating Documents
Documents are changed with methods such as update_one() and update_many(). An update normally contains a query filter and an update operator. The $set operator assigns values, while $inc increases or decreases a numeric field. The result reports how many documents matched and how many were modified.
Example: Update one product
# Update the Keyboard document.
result = products.update_one(
{"name": "Keyboard"},
{
"$set": {
"price": 44.99
},
"$inc": {
"quantity": 5
}
}
)
print("Matched:", result.matched_count)
print("Modified:", result.modified_count)
# Read the updated product.
product = products.find_one(
{"name": "Keyboard"},
{"_id": 0}
)
print(product)
Output:
Matched: 1
Modified: 1
{'name': 'Keyboard', 'price': 44.99, 'quantity': 15, 'available': True}
Output Explanation:
The price is replaced with 44.99, while the quantity is increased by five. One document matched and one document changed.
Example: Update several products
# Mark products with no inventory as unavailable.
result = products.update_many(
{"quantity": 0},
{
"$set": {
"available": False
}
}
)
print("Documents modified:", result.modified_count)
Possible Output:
Documents modified: 2
Output Explanation:
Every product with a quantity of zero is updated. The exact count depends on the documents stored in the collection.
43.9 Deleting Documents
MongoDB removes records with delete_one() or delete_many(). The query filter decides which documents are deleted. A broad or empty filter can remove many documents, so deletion conditions must be checked carefully. The deletion result includes a deleted_count property showing how many documents were removed.
Example: Delete one document
# Delete one product by name.
result = products.delete_one(
{"name": "Old Keyboard"}
)
print("Documents deleted:", result.deleted_count)
Possible Output:
Documents deleted: 1
Output Explanation:
MongoDB removes the first document matching the filter. A count of one confirms that a document was deleted.
Example: Delete several documents
# Delete discontinued products with no inventory.
result = products.delete_many(
{
"discontinued": True,
"quantity": 0
}
)
print("Documents deleted:", result.deleted_count)
Possible Output:
Documents deleted: 3
Output Explanation:
Only documents meeting both conditions are removed. The exact number depends on the current collection contents.
43.10 MongoDB Queries
MongoDB query filters use documents containing field names, values, and operators. Comparison operators include $gt, $gte, $lt, $lte, $ne, and $in. Conditions can be combined with logical operators. Dot notation can query fields stored inside nested documents.
Example: Price and inventory query
# Find available products below $100.
query = {
"price": {
"$lt": 100
},
"quantity": {
"$gt": 0
}
}
# Sort from highest to lowest price.
cursor = products.find(
query,
{
"_id": 0,
"name": 1,
"price": 1,
"quantity": 1
}
).sort("price", -1)
for product in cursor:
print(product)
Example Output:
{'name': 'Keyboard', 'price': 49.99, 'quantity': 10}
{'name': 'Mouse', 'price': 24.5, 'quantity': 25}
Output Explanation:
The query selects products priced below 100 with a quantity greater than zero. A sort value of negative one requests descending order.
Example: Query a nested field
# Find products with 16 GB of memory.
cursor = products.find(
{
"specifications.memory": "16 GB"
},
{
"_id": 0,
"name": 1,
"specifications.memory": 1
}
)
for product in cursor:
print(product)
Example Output:
{'name': 'Laptop', 'specifications': {'memory': '16 GB'}}
Output Explanation:
Dot notation reaches the memory field inside the nested specifications document.
43.11 MongoDB Aggregations
An aggregation pipeline processes documents through a sequence of stages. Each stage performs an operation such as filtering, grouping, calculating, sorting, reshaping, or limiting results. Common stages include $match, $group, $project, $sort, and $limit. The output of one stage becomes the input of the next.
Example: Calculate category totals
# Build an aggregation pipeline.
pipeline = [
# Include only completed orders.
{
"$match": {
"status": "completed"
}
},
# Group orders by category.
{
"$group": {
"_id": "$category",
"order_count": {
"$sum": 1
},
"total_sales": {
"$sum": "$total"
},
"average_sale": {
"$avg": "$total"
}
}
},
# Sort by total sales.
{
"$sort": {
"total_sales": -1
}
}
]
# Run the aggregation.
results = orders.aggregate(pipeline)
for result in results:
print(
result["_id"],
result["order_count"],
round(result["total_sales"], 2),
round(result["average_sale"], 2)
)
Example Output:
Electronics 4 540.50 135.12
Books 3 120.00 40.00
Clothing 2 95.50 47.75
Output Explanation:
Completed orders are grouped by category. MongoDB calculates the number of orders, total sales, and average sale for each category before sorting the groups.
Example: Calculate inventory value
pipeline = [
{
"$project": {
"_id": 0,
"name": 1,
"inventory_value": {
"$multiply": [
"$price",
"$quantity"
]
}
}
},
{
"$sort": {
"inventory_value": -1
}
}
]
for result in products.aggregate(pipeline):
print(
result["name"],
round(result["inventory_value"], 2)
)
Example Output:
Monitor 1512.0
Keyboard 499.9
Mouse 612.5
Output Explanation:
The projection creates a calculated field by multiplying each product’s price by its quantity.
43.12 MongoDB Indexes
An index helps MongoDB locate matching documents without examining every document in a collection. Indexes are useful for fields frequently used in filters, sorting, and uniqueness checks. However, every index uses storage and must be updated during writes. Applications should create indexes based on real query requirements instead of indexing every field.
Example: Create a unique index
# Import the ascending direction constant.
from pymongo import ASCENDING
# Create a unique email index.
index_name = customers.create_index(
[
("email", ASCENDING)
],
unique=True
)
print("Index:", index_name)
Output:
Index: email_1
Output Explanation:
The index supports email searches and prevents two documents from using the same email address.
Example: Create a compound index
# Import index directions.
from pymongo import ASCENDING, DESCENDING
# Create an index for customer order history.
index_name = orders.create_index(
[
("customer_id", ASCENDING),
("created_at", DESCENDING)
]
)
print("Index:", index_name)
Output:
Index: customer_id_1_created_at_-1
Output Explanation:
The compound index can support queries that filter by customer and sort that customer’s orders by newest date.
Example: List collection indexes
# Display every products index.
for index in products.list_indexes():
print(index["name"])
Possible Output:
_id_
name_1
category_1_price_-1
Output Explanation:
MongoDB automatically creates the _id_ index. The other index names represent additional indexes created for application queries.
43.13 Redis Introduction
Redis is a fast data store commonly used for caching, temporary sessions, counters, queues, rankings, and real-time application data. Redis stores values under keys and supports several specialized data structures. Much of its active data is kept in memory for fast access, while persistence options can be configured when longer-term storage is required.
Example: Common Redis uses
Redis key examples:
session:user:42
cache:product:100
page_views:home
queue:emails
leaderboard:math
cart:customer:15
Explanation:
session:user:42 - Temporary user session
cache:product:100 - Cached product data
page_views:home - Counter
queue:emails - Work queue
leaderboard:math - Sorted ranking
cart:customer:15 - Shopping-cart information
Output Explanation:
Clear key names help organize Redis information. Colons are commonly used as readable separators between parts of a key.
43.14 Connecting Python to Redis
Python applications commonly connect to Redis with the redis package, often called redis-py. A Redis client contains the server host, port, database number, credentials, and response settings. Setting decode_responses=True causes text responses to be returned as Python strings instead of raw byte values.
Example: Install redis-py
python -m pip install redis
Example: Connect to Redis
# Import the Redis package.
import redis
# Create a client for the local Redis server.
client = redis.Redis(
host="localhost",
port=6379,
db=0,
decode_responses=True
)
try:
# Confirm that the server responds.
connected = client.ping()
print("Connected:", connected)
finally:
# Close the client connection resources.
client.close()
Output:
Connected: True
Output Explanation:
A successful ping returns True. Redis normally uses port 6379 unless its server configuration specifies another port.
43.15 Key-Value Storage
The simplest Redis structure stores one value under one key. The set() method creates or replaces a value, while get() retrieves it. Redis can also increment numeric values, check whether keys exist, rename keys, and remove keys. Applications should use consistent key names to avoid accidental conflicts.
Example: Store and read values
# Import Redis.
import redis
client = redis.Redis(
host="localhost",
port=6379,
decode_responses=True
)
# Store string values.
client.set(
"application:name",
"Student Manager"
)
client.set(
"application:environment",
"development"
)
# Retrieve the values.
print(client.get("application:name"))
print(client.get("application:environment"))
client.close()
Output:
Student Manager
development
Output Explanation:
Each value is stored under a unique key. Calling get() with the same key returns the stored value.
Example: Create a counter
# Reset the page-view counter.
client.set("page_views:home", 0)
# Increase the counter three times.
client.incr("page_views:home")
client.incr("page_views:home")
client.incr("page_views:home")
print(client.get("page_views:home"))
Output:
3
Output Explanation:
Redis stores the numeric text value and changes it atomically with each incr() operation.
43.16 Caching
Caching stores frequently requested information in a fast temporary location. An application first checks the cache. When data exists, it is returned without repeating a slower database query. When it is missing, the application loads the information from the main database and saves a cached copy. This pattern is commonly called cache-aside.
Example: Cache-aside product lookup
# Import JSON for storing dictionary data.
import json
# Import Redis.
import redis
client = redis.Redis(
host="localhost",
port=6379,
decode_responses=True
)
# Simulate a slower database query.
def load_product_from_database(product_id):
print("Loading from database...")
return {
"id": product_id,
"name": "Keyboard",
"price": 49.99
}
# Return a product using the cache.
def get_product(product_id):
# Build a consistent cache key.
cache_key = f"cache:product:{product_id}"
# Check Redis first.
cached_data = client.get(cache_key)
if cached_data is not None:
print("Loading from cache...")
return json.loads(cached_data)
# Load from the main database.
product = load_product_from_database(
product_id
)
# Save the cached copy for 60 seconds.
client.set(
cache_key,
json.dumps(product),
ex=60
)
return product
# First call misses the cache.
print(get_product(1))
# Second call uses the cache.
print(get_product(1))
client.close()
Output:
Loading from database...
{'id': 1, 'name': 'Keyboard', 'price': 49.99}
Loading from cache...
{'id': 1, 'name': 'Keyboard', 'price': 49.99}
Output Explanation:
The first call loads the product from the main database and caches it. The second call finds the cached JSON and avoids repeating the database operation.
43.17 Expiration
Redis keys can receive an expiration time. When the time ends, Redis removes the key automatically. Expiration is useful for caches, login sessions, temporary verification information, and limited-time application state. Time-to-live values can be checked in seconds with ttl() or in milliseconds with pttl().
Example: Create a key with expiration
# Store a temporary value for 60 seconds.
client.set(
"session:user:42",
"active",
ex=60
)
# Display the value.
print(client.get("session:user:42"))
# Display the remaining lifetime.
print(
"Seconds remaining:",
client.ttl("session:user:42")
)
Example Output:
active
Seconds remaining: 60
Output Explanation:
The exact remaining time may be slightly below 60 by the time it is printed. Redis removes the key automatically after its lifetime ends.
Example: Add expiration to an existing key
# Create a key without expiration.
client.set(
"temporary:message",
"Hello"
)
# Add a 30-second lifetime.
client.expire(
"temporary:message",
30
)
print(
client.ttl("temporary:message")
)
Example Output:
30
Output Explanation:
The key originally had no expiration. The expire() command gives it a temporary lifetime.
43.18 Redis Data Structures
Redis supports more than simple strings. Lists store ordered values, sets store unique values, hashes store field-value pairs, and sorted sets associate members with numeric scores. Selecting the correct structure can simplify application code and improve performance. Each structure provides commands designed for its specific behavior.
Example: Redis list
# Remove any previous example list.
client.delete("queue:emails")
# Add jobs to the right side.
client.rpush(
"queue:emails",
"welcome@example.com"
)
client.rpush(
"queue:emails",
"receipt@example.com"
)
# Display the complete list.
print(
client.lrange(
"queue:emails",
0,
-1
)
)
# Remove one job from the left side.
print(
"Next job:",
client.lpop("queue:emails")
)
Output:
['welcome@example.com', 'receipt@example.com']
Next job: welcome@example.com
Output Explanation:
The list preserves order. The first value added becomes the first value removed by lpop().
Example: Redis set
# Remove the old set.
client.delete("course:python:students")
# Add unique student names.
client.sadd(
"course:python:students",
"Michael",
"Sara",
"Ali",
"Michael"
)
# Display the number of unique members.
print(
"Students:",
client.scard(
"course:python:students"
)
)
# Check membership.
print(
client.sismember(
"course:python:students",
"Sara"
)
)
Output:
Students: 3
True
Output Explanation:
Michael is added twice but stored only once because sets contain unique members. Sara is confirmed as a member.
Example: Redis hash
# Store several fields under one hash key.
client.hset(
"user:42",
mapping={
"name": "Michael",
"email": "michael@example.com",
"grade": "7"
}
)
# Read all hash fields.
user = client.hgetall("user:42")
print(user)
print(user["name"])
Output:
{'name': 'Michael', 'email': 'michael@example.com', 'grade': '7'}
Michael
Output Explanation:
A hash groups related fields under one Redis key. With decoded responses enabled, the fields and values are returned as strings.
Example: Redis sorted set
# Remove the old leaderboard.
client.delete("leaderboard:python")
# Add students with scores.
client.zadd(
"leaderboard:python",
{
"Michael": 92,
"Sara": 97,
"Ali": 85
}
)
# Read highest scores first.
leaders = client.zrevrange(
"leaderboard:python",
0,
-1,
withscores=True
)
for position, item in enumerate(
leaders,
start=1
):
name, score = item
print(
position,
name,
int(score)
)
Output:
1 Sara 97
2 Michael 92
3 Ali 85
Output Explanation:
Sorted-set members are ordered by score. The reverse-range command returns the highest score first.
43.19 NoSQL Best Practices
Reliable NoSQL applications require deliberate data design. Documents should follow predictable structures even when the database allows flexibility. Applications should validate input, create indexes for important queries, limit returned fields, protect credentials, monitor slow operations, use expiration for temporary values, and avoid storing the same important information in several uncontrolled places.
Example: Validate a MongoDB document
# Validate product information before insertion.
def validate_product(product):
# Require a non-empty product name.
name = product.get("name")
if not isinstance(name, str):
raise ValueError(
"Product name must be text."
)
if not name.strip():
raise ValueError(
"Product name cannot be empty."
)
# Require a non-negative price.
price = product.get("price")
if not isinstance(price, (int, float)):
raise ValueError(
"Product price must be numeric."
)
if price < 0:
raise ValueError(
"Product price cannot be negative."
)
# Require a non-negative quantity.
quantity = product.get("quantity")
if not isinstance(quantity, int):
raise ValueError(
"Product quantity must be an integer."
)
if quantity < 0:
raise ValueError(
"Product quantity cannot be negative."
)
return True
product = {
"name": "Keyboard",
"price": 49.99,
"quantity": 10
}
print(validate_product(product))
Output:
True
Output Explanation:
The product passes all checks. Validation prevents obviously incorrect information from reaching the database.
Example: Read credentials from the environment
# Import the os module.
import os
# Read protected configuration.
mongo_uri = os.getenv("MONGODB_URI")
redis_host = os.getenv(
"REDIS_HOST",
"localhost"
)
# Validate the required MongoDB address.
if not mongo_uri:
raise ValueError(
"MONGODB_URI is required."
)
# Display only non-sensitive information.
print("Redis host:", redis_host)
print("MongoDB configuration loaded.")
Output:
Redis host: localhost
MongoDB configuration loaded.
Output Explanation:
The complete MongoDB address is not printed because it may contain a username and password.
Important NoSQL practices
- Validate document fields before saving them.
- Use consistent collection and key naming.
- Create indexes for important MongoDB queries.
- Do not create unnecessary indexes.
- Return only fields required by the application.
- Use expiration for temporary Redis data.
- Handle unavailable database servers clearly.
- Never place passwords directly in public code.
- Use encrypted connections in production.
- Create backups and test recovery procedures.
43.20 Chapter Project
In this project, you will create a NoSQL product catalog using MongoDB for permanent product documents and Redis for fast product caching, page-view counters, and recently viewed products. The application demonstrates MongoDB CRUD operations, queries, aggregation, indexes, Redis key-value storage, hashes, lists, counters, expiration, validation, and safe environment configuration.
Step 1: Create the project structure
nosql_product_manager/
│
├── app.py
├── config.py
├── database.py
├── product_service.py
└── cache_service.py
Step 2: Install the required packages
python -m pip install pymongo redis
Step 3: Create config.py
# config.py
# Import environment-variable support.
import os
class Settings:
# Read the MongoDB connection address.
MONGODB_URI = os.getenv(
"MONGODB_URI",
"mongodb://localhost:27017/"
)
# Read the MongoDB database name.
MONGODB_DATABASE = os.getenv(
"MONGODB_DATABASE",
"product_manager"
)
# Read Redis connection settings.
REDIS_HOST = os.getenv(
"REDIS_HOST",
"localhost"
)
REDIS_PORT = int(
os.getenv("REDIS_PORT", "6379")
)
REDIS_DATABASE = int(
os.getenv("REDIS_DATABASE", "0")
)
REDIS_PASSWORD = os.getenv(
"REDIS_PASSWORD"
)
# Store product cache lifetime.
CACHE_SECONDS = int(
os.getenv("CACHE_SECONDS", "120")
)
def validate_settings():
# Confirm the MongoDB address exists.
if not Settings.MONGODB_URI.strip():
raise ValueError(
"MONGODB_URI cannot be empty."
)
# Confirm the database name exists.
if not Settings.MONGODB_DATABASE.strip():
raise ValueError(
"MONGODB_DATABASE cannot be empty."
)
# Confirm the Redis port is valid.
if not 1 <= Settings.REDIS_PORT <= 65535:
raise ValueError(
"REDIS_PORT is invalid."
)
# Confirm the cache lifetime is positive.
if Settings.CACHE_SECONDS <= 0:
raise ValueError(
"CACHE_SECONDS must be positive."
)
Step 4: Create database.py
# database.py
# Import MongoDB tools.
from pymongo import (
MongoClient,
ASCENDING,
DESCENDING
)
# Import Redis.
import redis
# Import application settings.
from config import Settings
# Create the MongoDB client.
mongo_client = MongoClient(
Settings.MONGODB_URI,
serverSelectionTimeoutMS=5000
)
# Select the application database.
mongo_database = mongo_client[
Settings.MONGODB_DATABASE
]
# Select the products collection.
products_collection = mongo_database[
"products"
]
# Create the Redis client.
redis_client = redis.Redis(
host=Settings.REDIS_HOST,
port=Settings.REDIS_PORT,
db=Settings.REDIS_DATABASE,
password=Settings.REDIS_PASSWORD,
decode_responses=True
)
# Test both database connections.
def test_connections():
# Test MongoDB.
mongo_client.admin.command("ping")
# Test Redis.
redis_client.ping()
# Create MongoDB indexes.
def create_indexes():
# Prevent duplicate SKUs.
products_collection.create_index(
[
("sku", ASCENDING)
],
unique=True
)
# Support category and price queries.
products_collection.create_index(
[
("category", ASCENDING),
("price", ASCENDING)
]
)
# Support inventory sorting.
products_collection.create_index(
[
("quantity", DESCENDING)
]
)
# Close both clients.
def close_connections():
mongo_client.close()
redis_client.close()
Step 5: Create cache_service.py
# cache_service.py
# Import JSON conversion support.
import json
# Import Redis and settings.
from database import redis_client
from config import Settings
# Create a cache key for a product.
def product_cache_key(product_id):
return f"cache:product:{product_id}"
# Cache one product document.
def cache_product(product):
# Convert ObjectId to text before JSON conversion.
cache_document = product.copy()
cache_document["_id"] = str(
cache_document["_id"]
)
# Store the product with expiration.
redis_client.set(
product_cache_key(
cache_document["_id"]
),
json.dumps(cache_document),
ex=Settings.CACHE_SECONDS
)
# Retrieve a cached product.
def get_cached_product(product_id):
cached_value = redis_client.get(
product_cache_key(product_id)
)
if cached_value is None:
return None
return json.loads(cached_value)
# Remove a product from the cache.
def remove_cached_product(product_id):
redis_client.delete(
product_cache_key(product_id)
)
# Increase the view counter.
def record_product_view(product_id):
counter_key = (
f"product:views:{product_id}"
)
# Increase the counter.
view_count = redis_client.incr(
counter_key
)
# Add the product to the recent-view list.
recent_key = "products:recently-viewed"
redis_client.lpush(
recent_key,
product_id
)
# Keep only the ten newest entries.
redis_client.ltrim(
recent_key,
0,
9
)
return view_count
# Return a product's view count.
def get_product_view_count(product_id):
value = redis_client.get(
f"product:views:{product_id}"
)
if value is None:
return 0
return int(value)
# Return recently viewed product IDs.
def get_recent_product_ids():
return redis_client.lrange(
"products:recently-viewed",
0,
9
)
Step 6: Create product_service.py
# product_service.py
# Import decimal conversion support.
from decimal import Decimal, InvalidOperation
# Import MongoDB ObjectId.
from bson import ObjectId
# Import duplicate-key errors.
from pymongo.errors import DuplicateKeyError
# Import the products collection.
from database import products_collection
# Import cache functions.
from cache_service import (
cache_product,
get_cached_product,
remove_cached_product,
record_product_view,
get_product_view_count,
get_recent_product_ids
)
# Convert a product to printable information.
def prepare_product(product):
if product is None:
return None
result = product.copy()
result["_id"] = str(result["_id"])
return result
# Validate product information.
def validate_product_data(
name,
sku,
category,
price,
quantity
):
# Clean text values.
name = name.strip()
sku = sku.strip().upper()
category = category.strip()
if not name:
raise ValueError(
"Product name cannot be empty."
)
if not sku:
raise ValueError(
"Product SKU cannot be empty."
)
if not category:
raise ValueError(
"Product category cannot be empty."
)
# Convert price safely.
try:
price = Decimal(str(price))
except InvalidOperation as error:
raise ValueError(
"Product price is invalid."
) from error
if price < 0:
raise ValueError(
"Product price cannot be negative."
)
if not isinstance(quantity, int):
raise ValueError(
"Product quantity must be an integer."
)
if quantity < 0:
raise ValueError(
"Product quantity cannot be negative."
)
return {
"name": name,
"sku": sku,
"category": category,
"price": float(price),
"quantity": quantity,
"available": quantity > 0
}
# Add a new MongoDB product.
def add_product(
name,
sku,
category,
price,
quantity
):
product = validate_product_data(
name,
sku,
category,
price,
quantity
)
result = products_collection.insert_one(
product
)
return str(result.inserted_id)
# Return one product using cache-aside.
def get_product(product_id):
# Check Redis first.
cached_product = get_cached_product(
product_id
)
if cached_product is not None:
source = "Redis cache"
product = cached_product
else:
# Validate the MongoDB identifier.
if not ObjectId.is_valid(product_id):
raise ValueError(
"Product ID is invalid."
)
# Load the product from MongoDB.
product = products_collection.find_one(
{
"_id": ObjectId(product_id)
}
)
if product is None:
return None, "Not found", 0
# Save the product in Redis.
cache_product(product)
# Prepare the ObjectId for printing.
product = prepare_product(product)
source = "MongoDB"
# Record this product view.
view_count = record_product_view(
product_id
)
return product, source, view_count
# Return products with optional filters.
def list_products(
category=None,
maximum_price=None,
available_only=False
):
query = {}
if category:
query["category"] = category.strip()
if maximum_price is not None:
query["price"] = {
"$lte": float(maximum_price)
}
if available_only:
query["quantity"] = {
"$gt": 0
}
cursor = products_collection.find(
query
).sort(
[
("category", 1),
("name", 1)
]
)
return [
prepare_product(product)
for product in cursor
]
# Update a product.
def update_product(
product_id,
name,
category,
price,
quantity
):
if not ObjectId.is_valid(product_id):
raise ValueError(
"Product ID is invalid."
)
# Use the existing SKU while validating new fields.
current = products_collection.find_one(
{
"_id": ObjectId(product_id)
}
)
if current is None:
return False
updated_product = validate_product_data(
name,
current["sku"],
category,
price,
quantity
)
# Do not change the SKU.
updated_product.pop("sku")
result = products_collection.update_one(
{
"_id": ObjectId(product_id)
},
{
"$set": updated_product
}
)
# Remove outdated cached information.
remove_cached_product(product_id)
return result.matched_count == 1
# Delete one product.
def delete_product(product_id):
if not ObjectId.is_valid(product_id):
raise ValueError(
"Product ID is invalid."
)
result = products_collection.delete_one(
{
"_id": ObjectId(product_id)
}
)
# Remove the cached product.
remove_cached_product(product_id)
return result.deleted_count == 1
# Return category inventory statistics.
def get_category_statistics():
pipeline = [
{
"$group": {
"_id": "$category",
"product_count": {
"$sum": 1
},
"total_quantity": {
"$sum": "$quantity"
},
"average_price": {
"$avg": "$price"
},
"inventory_value": {
"$sum": {
"$multiply": [
"$price",
"$quantity"
]
}
}
}
},
{
"$sort": {
"inventory_value": -1
}
}
]
return list(
products_collection.aggregate(
pipeline
)
)
# Return recently viewed products.
def get_recently_viewed_products():
product_ids = get_recent_product_ids()
products = []
# Avoid displaying duplicate recent IDs.
seen = set()
for product_id in product_ids:
if product_id in seen:
continue
seen.add(product_id)
if not ObjectId.is_valid(product_id):
continue
product = products_collection.find_one(
{
"_id": ObjectId(product_id)
}
)
if product is None:
continue
prepared = prepare_product(product)
prepared["view_count"] = (
get_product_view_count(
product_id
)
)
products.append(prepared)
return products
Step 7: Create app.py
# app.py
# Import database exceptions.
from pymongo.errors import (
DuplicateKeyError,
PyMongoError
)
from redis.exceptions import RedisError
# Import configuration.
from config import validate_settings
# Import database management.
from database import (
test_connections,
create_indexes,
close_connections
)
# Import product operations.
from product_service import (
add_product,
get_product,
list_products,
update_product,
delete_product,
get_category_statistics,
get_recently_viewed_products
)
# Display the main menu.
def show_menu():
print("\nNoSQL Product Manager")
print("1. Add product")
print("2. View product")
print("3. List products")
print("4. Search products")
print("5. Update product")
print("6. Delete product")
print("7. Category statistics")
print("8. Recently viewed products")
print("9. Exit")
# Display one product.
def display_product(product):
print("ID:", product["_id"])
print("Name:", product["name"])
print("SKU:", product["sku"])
print("Category:", product["category"])
print(f'Price: ${product["price"]:.2f}')
print("Quantity:", product["quantity"])
print("Available:", product["available"])
# Display several products.
def display_products(products):
if not products:
print("No products were found.")
return
for product in products:
print(
f'{product["_id"]} | '
f'{product["name"]} | '
f'{product["sku"]} | '
f'{product["category"]} | '
f'${product["price"]:.2f} | '
f'Quantity: {product["quantity"]}'
)
# Run the application.
def main():
# Validate configuration.
validate_settings()
# Confirm both servers are available.
test_connections()
# Create required MongoDB indexes.
create_indexes()
print(
"MongoDB and Redis connections "
"are ready."
)
while True:
show_menu()
choice = input(
"Choose an option: "
).strip()
try:
if choice == "1":
name = input(
"Product name: "
)
sku = input(
"Product SKU: "
)
category = input(
"Category: "
)
price = input(
"Price: "
)
quantity = int(
input("Quantity: ")
)
product_id = add_product(
name,
sku,
category,
price,
quantity
)
print(
"Product created with ID:",
product_id
)
elif choice == "2":
product_id = input(
"Product ID: "
).strip()
product, source, views = (
get_product(product_id)
)
if product is None:
print("Product not found.")
continue
display_product(product)
print("Loaded from:", source)
print("Views:", views)
elif choice == "3":
products = list_products()
display_products(products)
elif choice == "4":
category = input(
"Category or leave empty: "
).strip()
maximum_price_text = input(
"Maximum price or leave empty: "
).strip()
available_text = input(
"Available products only? "
"(yes/no): "
).strip().lower()
maximum_price = (
float(maximum_price_text)
if maximum_price_text
else None
)
available_only = (
available_text == "yes"
)
products = list_products(
category=(
category
if category
else None
),
maximum_price=maximum_price,
available_only=available_only
)
display_products(products)
elif choice == "5":
product_id = input(
"Product ID: "
).strip()
name = input(
"New product name: "
)
category = input(
"New category: "
)
price = input(
"New price: "
)
quantity = int(
input("New quantity: ")
)
updated = update_product(
product_id,
name,
category,
price,
quantity
)
if updated:
print(
"Product updated."
)
else:
print(
"Product not found."
)
elif choice == "6":
product_id = input(
"Product ID: "
).strip()
deleted = delete_product(
product_id
)
if deleted:
print(
"Product deleted."
)
else:
print(
"Product not found."
)
elif choice == "7":
statistics = (
get_category_statistics()
)
if not statistics:
print(
"No statistics are available."
)
for item in statistics:
print(
f'Category: {item["_id"]} | '
f'Products: '
f'{item["product_count"]} | '
f'Quantity: '
f'{item["total_quantity"]} | '
f'Average price: '
f'${item["average_price"]:.2f} | '
f'Inventory value: '
f'${item["inventory_value"]:.2f}'
)
elif choice == "8":
products = (
get_recently_viewed_products()
)
if not products:
print(
"No recently viewed products."
)
for product in products:
print(
f'{product["name"]} | '
f'Views: '
f'{product["view_count"]}'
)
elif choice == "9":
print("Goodbye.")
break
else:
print("Invalid option.")
except ValueError as error:
print("Input error:", error)
except DuplicateKeyError:
print(
"A product with that SKU "
"already exists."
)
except PyMongoError as error:
print(
"A MongoDB operation failed."
)
print(
"Error type:",
type(error).__name__
)
except RedisError as error:
print(
"A Redis operation failed."
)
print(
"Error type:",
type(error).__name__
)
# Start the program.
if __name__ == "__main__":
try:
main()
finally:
close_connections()
Step 8: Optional environment variables
MONGODB_URI=mongodb://localhost:27017/
MONGODB_DATABASE=product_manager
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DATABASE=0
CACHE_SECONDS=120
Step 9: Run the project
python app.py
Example Output:
MongoDB and Redis connections are ready.
NoSQL Product Manager
1. Add product
2. View product
3. List products
4. Search products
5. Update product
6. Delete product
7. Category statistics
8. Recently viewed products
9. Exit
Choose an option: 1
Product name: Keyboard
Product SKU: KB-100
Category: Electronics
Price: 49.99
Quantity: 10
Product created with ID: 669b780c760595af04a12001
Choose an option: 1
Product name: Mouse
Product SKU: MS-200
Category: Electronics
Price: 24.50
Quantity: 25
Product created with ID: 669b780c760595af04a12002
Choose an option: 2
Product ID: 669b780c760595af04a12001
ID: 669b780c760595af04a12001
Name: Keyboard
SKU: KB-100
Category: Electronics
Price: $49.99
Quantity: 10
Available: True
Loaded from: MongoDB
Views: 1
Choose an option: 2
Product ID: 669b780c760595af04a12001
ID: 669b780c760595af04a12001
Name: Keyboard
SKU: KB-100
Category: Electronics
Price: $49.99
Quantity: 10
Available: True
Loaded from: Redis cache
Views: 2
Choose an option: 7
Category: Electronics | Products: 2 | Quantity: 35 | Average price: $37.24 | Inventory value: $1112.40
Choose an option: 8
Keyboard | Views: 2
Output Explanation:
MongoDB permanently stores product documents and performs queries, updates, deletions, aggregations, and indexed lookups. Redis temporarily caches product details, counts views, and maintains a recently viewed list. The first product lookup comes from MongoDB, while the second lookup uses the Redis cache.
Project Summary
This project demonstrates MongoDB connections, collections, documents, CRUD operations, query filters, sorting, aggregation pipelines, unique indexes, compound indexes, ObjectId validation, Redis connections, strings, counters, lists, JSON caching, expiration, cache invalidation, environment configuration, validation, and database error handling.
