380 lines
14 KiB
Python
380 lines
14 KiB
Python
# Imports the os module so the app can read environment variables
|
|
# such as SECRET_KEY, MONGO_URI, and PORT.
|
|
import os
|
|
|
|
# Imports time so the code can wait and retry when MongoDB is not ready yet.
|
|
import time
|
|
|
|
# Imports datetime so the app can store the current date and time
|
|
# when a new product is created.
|
|
from datetime import datetime
|
|
|
|
# Flask imports:
|
|
# - Flask creates the web application
|
|
# - render_template loads HTML files from the templates folder
|
|
# - request reads form data sent from the browser
|
|
# - redirect sends the user to another page
|
|
# - url_for builds a URL from a route function name
|
|
# - flash stores short success/error messages to show on the next page
|
|
from flask import Flask, render_template, request, redirect, url_for, flash
|
|
|
|
# PyMongo imports:
|
|
# - MongoClient connects Python to MongoDB
|
|
# - ASCENDING is used when creating an index in ascending order
|
|
from pymongo import MongoClient, ASCENDING
|
|
|
|
# Import the specific error raised when trying to insert a duplicate value
|
|
# into a field that has a unique index.
|
|
from pymongo.errors import DuplicateKeyError
|
|
|
|
# Imports ObjectId so MongoDB document IDs in the URL can be converted
|
|
# into the correct MongoDB ID type for queries.
|
|
from bson.objectid import ObjectId
|
|
|
|
# Creates the Flask application object.
|
|
app = Flask(__name__)
|
|
|
|
# Sets the secret key used by Flask for flash messages and sessions.
|
|
# It first tries to read SECRET_KEY from the environment.
|
|
# If not found, it uses "dev-secret-key" as a default.
|
|
app.secret_key = os.getenv("SECRET_KEY", "dev-secret-key")
|
|
|
|
# Reads the MongoDB connection string from an environment variable.
|
|
# If none is provided, it uses the default Docker service connection.
|
|
MONGO_URI = os.getenv("MONGO_URI", "mongodb://admin:admin123@mongo:27017/")
|
|
|
|
# Reads the MongoDB database name from the environment.
|
|
# Defaults to "appdb".
|
|
MONGO_DB_NAME = os.getenv("MONGO_DB_NAME", "appdb")
|
|
|
|
# Reads the MongoDB collection name from the environment.
|
|
# Defaults to "products".
|
|
MONGO_COLLECTION_NAME = os.getenv("MONGO_COLLECTION_NAME", "products")
|
|
|
|
# Reads the port number for the Flask app from the environment.
|
|
# Defaults to 8080 and converts it into an integer.
|
|
PORT = int(os.getenv("PORT", "8080"))
|
|
|
|
# These variables will be filled later after MongoDB is connected.
|
|
# mongo_client will store the MongoClient object.
|
|
# products_collection will store the products collection object.
|
|
mongo_client = None
|
|
products_collection = None
|
|
|
|
|
|
# This function connects the application to MongoDB.
|
|
# It retries multiple times because, in Docker, MongoDB may take a few
|
|
# seconds to start after the Flask app container starts.
|
|
def connect_to_mongo():
|
|
# Tells Python that these names refer to the global variables above,
|
|
# not new local variables inside the function.
|
|
global mongo_client, products_collection
|
|
|
|
# Try up to 30 times to connect to MongoDB.
|
|
for attempt in range(30):
|
|
try:
|
|
# Create the MongoDB client.
|
|
# serverSelectionTimeoutMS=2000 means each failed attempt times out
|
|
# after 2 seconds instead of hanging too long.
|
|
mongo_client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=2000)
|
|
|
|
# Ping the database to confirm it is reachable.
|
|
mongo_client.admin.command("ping")
|
|
|
|
# Select the database configured in MONGO_DB_NAME.
|
|
db = mongo_client[MONGO_DB_NAME]
|
|
|
|
# Select the collection configured in MONGO_COLLECTION_NAME.
|
|
products_collection = db[MONGO_COLLECTION_NAME]
|
|
|
|
# Create a unique index on product_id.
|
|
# This prevents two products from having the same product_id.
|
|
products_collection.create_index(
|
|
[("product_id", ASCENDING)],
|
|
unique=True
|
|
)
|
|
|
|
# If connection and setup succeed, exit the function.
|
|
return
|
|
|
|
except Exception as e:
|
|
# If connection fails, print a message showing which retry attempt failed.
|
|
print(f"MongoDB not ready yet (attempt {attempt + 1}/30): {e}")
|
|
|
|
# Wait 2 seconds before trying again.
|
|
time.sleep(2)
|
|
|
|
# If all 30 attempts fail, stop the app with an error.
|
|
raise RuntimeError("Could not connect to MongoDB after multiple attempts.")
|
|
|
|
|
|
# Converts a form value into a float.
|
|
# Used for fields like price and rating.
|
|
def parse_float(value):
|
|
# If value is None, replace it with an empty string, then strip spaces.
|
|
value = (value or "").strip()
|
|
|
|
# If the field was left empty, return None instead of crashing.
|
|
if value == "":
|
|
return None
|
|
|
|
# Convert the value to a float.
|
|
return float(value)
|
|
|
|
|
|
# Converts a form value into an integer.
|
|
# Used for product_id.
|
|
def parse_int(value):
|
|
# If value is None, replace it with an empty string, then strip spaces.
|
|
value = (value or "").strip()
|
|
|
|
# If the field was left empty, return None.
|
|
if value == "":
|
|
return None
|
|
|
|
# Convert the value to an integer.
|
|
return int(value)
|
|
|
|
|
|
# Converts a comma-separated string of tags into a Python list.
|
|
# Example: "gaming, pc, accessories" becomes ["gaming", "pc", "accessories"].
|
|
def parse_tags(value):
|
|
# Replace None with empty string and trim spaces.
|
|
value = (value or "").strip()
|
|
|
|
# If the field is empty, return None.
|
|
if not value:
|
|
return None
|
|
|
|
# Split on commas, trim spaces around each tag, and discard empty tags.
|
|
return [tag.strip() for tag in value.split(",") if tag.strip()]
|
|
|
|
|
|
# Converts a raw MongoDB document into a cleaner dictionary for HTML templates.
|
|
# This makes values easier to display and avoids issues with ObjectId or None values.
|
|
def normalize_product(product):
|
|
return {
|
|
# Convert MongoDB's ObjectId into a normal string so it can be used in URLs.
|
|
"_id": str(product.get("_id")),
|
|
|
|
# Read product_id, defaulting to empty string if missing.
|
|
"product_id": product.get("product_id", ""),
|
|
|
|
# Read product name.
|
|
"name": product.get("name", ""),
|
|
|
|
# Read category.
|
|
"category": product.get("category", ""),
|
|
|
|
# If price is None, use empty string so the form field looks blank.
|
|
# Otherwise keep the real price value.
|
|
"price": "" if product.get("price") is None else product.get("price"),
|
|
|
|
# Make sure in_stock is always a real boolean.
|
|
"in_stock": bool(product.get("in_stock", False)),
|
|
|
|
# If rating is None, use empty string for cleaner display.
|
|
"rating": "" if product.get("rating") is None else product.get("rating"),
|
|
|
|
# Always return tags as a list.
|
|
"tags": product.get("tags") or [],
|
|
|
|
# Also create a comma-separated version of tags for text input fields.
|
|
"tags_text": ", ".join(product.get("tags") or []),
|
|
|
|
# Format created_at as a readable date-time string if it exists.
|
|
"created_at": product.get("created_at").strftime("%Y-%m-%d %H:%M:%S")
|
|
if product.get("created_at") else "",
|
|
|
|
# Read supplier name and country.
|
|
"supplier_name": product.get("supplier_name", ""),
|
|
"supplier_country": product.get("supplier_country", "")
|
|
}
|
|
|
|
|
|
# This route handles visits to the home page "/".
|
|
# It loads all products from MongoDB and shows them in index.html.
|
|
@app.route("/")
|
|
def index():
|
|
# Find all documents in the collection and sort them by product_id ascending.
|
|
raw_products = list(products_collection.find().sort("product_id", 1))
|
|
|
|
# Normalize every MongoDB document before sending it to the template.
|
|
products = [normalize_product(p) for p in raw_products]
|
|
|
|
# Render the HTML page and pass the products list into it.
|
|
return render_template("index.html", products=products)
|
|
|
|
|
|
# This route handles form submissions for creating a new product.
|
|
# It only accepts POST requests.
|
|
@app.route("/create", methods=["POST"])
|
|
def create_product():
|
|
try:
|
|
# Build a new product dictionary from submitted form values.
|
|
product = {
|
|
# Convert product_id to integer.
|
|
"product_id": parse_int(request.form.get("product_id")),
|
|
|
|
# Read the required name field and trim spaces.
|
|
"name": request.form.get("name", "").strip(),
|
|
|
|
# Optional text fields become None if left blank.
|
|
"category": request.form.get("category", "").strip() or None,
|
|
|
|
# Convert numeric fields.
|
|
"price": parse_float(request.form.get("price")),
|
|
|
|
# Checkbox is "on" when checked.
|
|
"in_stock": request.form.get("in_stock") == "on",
|
|
|
|
# Convert rating to float.
|
|
"rating": parse_float(request.form.get("rating")),
|
|
|
|
# Convert tags from text to list.
|
|
"tags": parse_tags(request.form.get("tags")),
|
|
|
|
# Store the current UTC time as the creation time.
|
|
"created_at": datetime.utcnow(),
|
|
|
|
# Optional supplier fields.
|
|
"supplier_name": request.form.get("supplier_name", "").strip() or None,
|
|
"supplier_country": request.form.get("supplier_country", "").strip() or None
|
|
}
|
|
|
|
# Validate that product_id was provided.
|
|
if not product["product_id"]:
|
|
flash("Product ID is required.", "error")
|
|
return redirect(url_for("index"))
|
|
|
|
# Validate that name was provided.
|
|
if not product["name"]:
|
|
flash("Name is required.", "error")
|
|
return redirect(url_for("index"))
|
|
|
|
# Insert the new product into MongoDB.
|
|
products_collection.insert_one(product)
|
|
|
|
# Store a success message to show on the next page load.
|
|
flash("Product created.", "success")
|
|
|
|
except DuplicateKeyError:
|
|
# Happens when another product already has the same product_id.
|
|
flash("That product_id already exists.", "error")
|
|
|
|
except ValueError:
|
|
# Happens when user enters invalid numbers in numeric fields.
|
|
flash("Invalid number entered for product_id, price, or rating.", "error")
|
|
|
|
except Exception as e:
|
|
# Catch-all for any other error.
|
|
flash(f"Create failed: {e}", "error")
|
|
|
|
# Return the user to the home page after processing.
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
# This route opens the edit page for a specific product.
|
|
# The product's MongoDB _id is passed in the URL.
|
|
@app.route("/edit/<product_id>")
|
|
def edit_product(product_id):
|
|
try:
|
|
# Look up the product by its MongoDB ObjectId.
|
|
product = products_collection.find_one({"_id": ObjectId(product_id)})
|
|
|
|
# If no product is found, show an error and go back home.
|
|
if not product:
|
|
flash("Product not found.", "error")
|
|
return redirect(url_for("index"))
|
|
|
|
# Render edit.html with the normalized product data.
|
|
return render_template("edit.html", product=normalize_product(product))
|
|
|
|
except Exception:
|
|
# Happens if product_id is not a valid ObjectId or something else fails.
|
|
flash("Invalid product id.", "error")
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
# This route handles updating an existing product.
|
|
# It accepts POST data from the edit form.
|
|
@app.route("/update/<product_id>", methods=["POST"])
|
|
def update_product(product_id):
|
|
try:
|
|
# Build a dictionary of updated field values from the form.
|
|
updated_product = {
|
|
"name": request.form.get("name", "").strip(),
|
|
"category": request.form.get("category", "").strip() or None,
|
|
"price": parse_float(request.form.get("price")),
|
|
"in_stock": request.form.get("in_stock") == "on",
|
|
"rating": parse_float(request.form.get("rating")),
|
|
"tags": parse_tags(request.form.get("tags")),
|
|
"supplier_name": request.form.get("supplier_name", "").strip() or None,
|
|
"supplier_country": request.form.get("supplier_country", "").strip() or None
|
|
}
|
|
|
|
# Require the name field.
|
|
if not updated_product["name"]:
|
|
flash("Name is required.", "error")
|
|
return redirect(url_for("edit_product", product_id=product_id))
|
|
|
|
# Update the matching document in MongoDB.
|
|
# $set changes only the listed fields instead of replacing the whole document.
|
|
result = products_collection.update_one(
|
|
{"_id": ObjectId(product_id)},
|
|
{"$set": updated_product}
|
|
)
|
|
|
|
# If no document matched the given ID, show an error.
|
|
if result.matched_count == 0:
|
|
flash("Product not found.", "error")
|
|
else:
|
|
# Otherwise report success.
|
|
flash("Product updated.", "success")
|
|
|
|
except ValueError:
|
|
# Numeric parsing failed.
|
|
flash("Invalid number entered for price or rating.", "error")
|
|
|
|
except Exception as e:
|
|
# Any other problem during update.
|
|
flash(f"Update failed: {e}", "error")
|
|
|
|
# Return to the main page after update.
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
# This route deletes a product.
|
|
# It accepts POST requests so deletion is triggered by a form submission.
|
|
@app.route("/delete/<product_id>", methods=["POST"])
|
|
def delete_product(product_id):
|
|
try:
|
|
# Delete the document whose _id matches the URL parameter.
|
|
result = products_collection.delete_one({"_id": ObjectId(product_id)})
|
|
|
|
# If nothing was deleted, the product was not found.
|
|
if result.deleted_count == 0:
|
|
flash("Product not found.", "error")
|
|
else:
|
|
# Otherwise show a success message.
|
|
flash("Product deleted.", "success")
|
|
|
|
except Exception as e:
|
|
# Handle invalid ObjectId or any other delete error.
|
|
flash(f"Delete failed: {e}", "error")
|
|
|
|
# Return the user to the home page.
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
# This block only runs when this file is executed directly.
|
|
# It will not run if the app is imported from another file.
|
|
if __name__ == "__main__":
|
|
# Connect to MongoDB before starting the web server.
|
|
connect_to_mongo()
|
|
|
|
# Start the Flask development server.
|
|
# host="0.0.0.0" makes it accessible from outside the container.
|
|
# port=PORT uses the configured port.
|
|
# debug=False disables Flask debug mode.
|
|
app.run(host="0.0.0.0", port=PORT, debug=False) |