EASYTUTORGUIDE

Practical tutorials, tools, courses, digital skills, and business promotion.

Free Learning
EasyTutorGuide Logo Learn, Build, & Grow

JavaScript – Chapter 2: Development Environment

Learn what JavaScript is, where it runs, why it is important, and how beginners can start using it with simple examples and clear output.

Beginner Friendly JavaScript Basics Web Development Code Examples
JavaScript Lesson 2 Chapter 2 Topics Day / Night Mode

Main reading content

python-course-chapter-41

Chapter 41: PostgreSQL and MySQL

Learn how Python connects to PostgreSQL and MySQL database servers, performs database operations, manages transactions, uses connection pools, and follows production database practices.

Goal: Build a complete Python application that safely connects to PostgreSQL or MySQL, manages related records, uses parameterized queries, handles errors, and follows reliable production practices.

Chapter 41 Topics

41.1 Introduction to Database Servers

A database server is a program that manages databases and accepts requests from applications over a network. Unlike SQLite, which stores data in a local file, PostgreSQL and MySQL normally run as separate services. Many users and applications can connect at the same time. The server manages security, users, permissions, transactions, backups, performance, and concurrent access.

Example: Information needed to connect to a server

# Store common database connection settings.
database_settings = {
    "host": "localhost",
    "port": 5432,
    "database": "school",
    "user": "app_user",
    "password": "example-password"
}

# Display the non-sensitive connection information.
print("Host:", database_settings["host"])
print("Port:", database_settings["port"])
print("Database:", database_settings["database"])
print("User:", database_settings["user"])
Output: 
Host: localhost
Port: 5432
Database: school
User: app_user

Explanation: A database client usually needs the server address, port, database name, username, and password. The password should not normally be printed or stored directly in public source code.

41.2 PostgreSQL

PostgreSQL is a powerful open-source relational database system. It supports standard SQL, transactions, foreign keys, views, advanced indexes, JSON data, stored functions, and many other features. PostgreSQL is commonly used for web applications, business systems, analytics, geographic information, and applications that require strong data integrity and advanced database capabilities.

Example: PostgreSQL table definition

CREATE TABLE customers (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(255) NOT NULL UNIQUE,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
Output: 
CREATE TABLE

Explanation: PostgreSQL creates a customers table with an automatically generated ID. The email must be unique, and the creation date is added automatically when a new customer is inserted.

41.3 MySQL

MySQL is a widely used relational database server. It is commonly used with websites, content-management systems, online stores, business applications, and cloud services. MySQL supports tables, relationships, transactions, indexes, views, stored procedures, and replication. Its SQL syntax is similar to PostgreSQL, although some data types, functions, and server features are different.

Example: MySQL table definition

CREATE TABLE products (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(150) NOT NULL,
    price DECIMAL(10, 2) NOT NULL,
    quantity INT NOT NULL DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
Output: 
Table 'products' created successfully.

Explanation: MySQL uses AUTO_INCREMENT to generate product IDs. The price supports two decimal places, and the quantity begins at zero when no value is provided.

41.4 Installing Database Drivers

Python needs a database driver to communicate with PostgreSQL or MySQL. A driver translates Python method calls and values into commands understood by the database server. PostgreSQL applications commonly use Psycopg, while MySQL applications may use MySQL Connector/Python. Drivers should be installed inside the project’s virtual environment so dependencies remain organized.

Example: Install PostgreSQL and MySQL drivers

# Install the PostgreSQL driver.
python -m pip install psycopg[binary]

# Install the official MySQL Connector/Python driver.
python -m pip install mysql-connector-python

Example: Confirm the drivers can be imported

# Import the PostgreSQL driver.
import psycopg

# Import the MySQL driver.
import mysql.connector

print("PostgreSQL driver imported.")
print("MySQL driver imported.")
Output: 
PostgreSQL driver imported.
MySQL driver imported.

Explanation: Successful imports show that the required drivers are installed in the current Python environment. The database servers must still be installed, running, and accessible before a connection can succeed.

41.5 Connecting to PostgreSQL

A PostgreSQL connection is created with the Psycopg connect() function. The application supplies the server host, port, database name, username, and password. After connecting, the program can create cursors and execute SQL. Connections should be closed after use, and sensitive credentials should be loaded from environment variables instead of being written directly in code.

Example: Connect to PostgreSQL

# Import the PostgreSQL driver.
import psycopg

# Open a PostgreSQL connection.
connection = psycopg.connect(
    host="localhost",
    port=5432,
    dbname="school",
    user="app_user",
    password="example-password"
)

# Confirm the connection.
print("Connected to PostgreSQL.")

# Close the connection.
connection.close()

print("Connection closed.")
Output: 
Connected to PostgreSQL.
Connection closed.

Explanation: The connection opens a session with the PostgreSQL server. The example closes it immediately, but a real application would execute SQL operations before closing.

41.6 Connecting to MySQL

MySQL Connector/Python creates connections with mysql.connector.connect(). The connection settings are similar to PostgreSQL, but the database-name parameter is commonly called database. The application should check whether the connection succeeded, perform its work, and close the connection in a finally block or another reliable cleanup structure.

Example: Connect to MySQL

# Import the MySQL driver.
import mysql.connector

# Open a MySQL connection.
connection = mysql.connector.connect(
    host="localhost",
    port=3306,
    database="store",
    user="app_user",
    password="example-password"
)

# Confirm that the connection is active.
if connection.is_connected():
    print("Connected to MySQL.")

# Close the connection.
connection.close()

print("Connection closed.")
Output: 
Connected to MySQL.
Connection closed.

Explanation: MySQL normally listens on port 3306. The is_connected() method checks whether the driver has an active connection to the database server.

41.7 Connection Strings

A connection string stores database connection information in one formatted value. It may contain the database system, username, password, host, port, and database name. Connection strings are useful for configuration and deployment, but passwords may contain special characters that require encoding. A complete connection string must never be printed in logs or exposed publicly.

Example: PostgreSQL connection string

# Import the PostgreSQL driver.
import psycopg

# Store the connection information.
connection_string = (
    "host=localhost "
    "port=5432 "
    "dbname=school "
    "user=app_user "
    "password=example-password"
)

# Connect using the string.
connection = psycopg.connect(connection_string)

print("PostgreSQL connection created.")

connection.close()

Example: URL-style database address

postgresql://app_user:password@localhost:5432/school

mysql://app_user:password@localhost:3306/store
Output: 
PostgreSQL connection created.

Explanation: The connection string combines several settings. In a real project, it should usually be loaded from a protected environment variable such as DATABASE_URL.

41.8 Database Cursors

A cursor is an object used to execute SQL statements and retrieve query results. The cursor sends commands through an open database connection. Methods such as execute(), fetchone(), and fetchall() perform common operations. Cursors should be closed when no longer needed, or managed with a context manager when the driver supports it.

Example: Use a PostgreSQL cursor

# Import the PostgreSQL driver.
import psycopg

# Open a connection using a context manager.
with psycopg.connect(
    host="localhost",
    dbname="school",
    user="app_user",
    password="example-password"
) as connection:

    # Create and manage a cursor.
    with connection.cursor() as cursor:
        # Ask PostgreSQL for its current date.
        cursor.execute("SELECT CURRENT_DATE")

        # Retrieve one row.
        row = cursor.fetchone()

        # Display the returned date.
        print("Database date:", row[0])
Example Output: 
Database date: 2026-07-19

Explanation: The cursor executes the query and returns one tuple. The first value in the tuple contains the date reported by the database server.

41.9 CRUD Operations

CRUD represents Create, Read, Update, and Delete. These are the four basic operations performed on stored information. SQL uses INSERT to create rows, SELECT to read them, UPDATE to change them, and DELETE to remove them. Python programs should perform these operations with parameterized queries and controlled transactions.

Example: PostgreSQL CRUD operations

# Import the PostgreSQL driver.
import psycopg

# Open the database connection.
with psycopg.connect(
    host="localhost",
    dbname="store",
    user="app_user",
    password="example-password"
) as connection:

    with connection.cursor() as cursor:
        # CREATE: Insert a new product.
        cursor.execute(
            """
            INSERT INTO products (name, price, quantity)
            VALUES (%s, %s, %s)
            RETURNING id
            """,
            ("Keyboard", 49.99, 10)
        )

        product_id = cursor.fetchone()[0]

        # READ: Retrieve the inserted product.
        cursor.execute(
            """
            SELECT id, name, price, quantity
            FROM products
            WHERE id = %s
            """,
            (product_id,)
        )

        print("Created:", cursor.fetchone())

        # UPDATE: Change the quantity.
        cursor.execute(
            """
            UPDATE products
            SET quantity = %s
            WHERE id = %s
            """,
            (15, product_id)
        )

        # READ: Retrieve the updated product.
        cursor.execute(
            """
            SELECT id, name, price, quantity
            FROM products
            WHERE id = %s
            """,
            (product_id,)
        )

        print("Updated:", cursor.fetchone())

        # DELETE: Remove the product.
        cursor.execute(
            """
            DELETE FROM products
            WHERE id = %s
            """,
            (product_id,)
        )

        print("Deleted rows:", cursor.rowcount)
Example Output: 
Created: (1, 'Keyboard', Decimal('49.99'), 10)
Updated: (1, 'Keyboard', Decimal('49.99'), 15)
Deleted rows: 1

Explanation: The program inserts, reads, updates, and deletes one product. PostgreSQL uses %s placeholders with Psycopg, while the values are supplied separately.

41.10 Transactions

A transaction groups several database operations into one logical unit. A successful transaction is committed, while a failed transaction is rolled back. Transactions protect data when multiple changes must succeed together. A money transfer, for example, must subtract from one account and add to another without saving only one side of the transfer.

Example: PostgreSQL account transfer

# Import the PostgreSQL driver.
import psycopg

try:
    # Open a connection.
    with psycopg.connect(
        host="localhost",
        dbname="bank",
        user="app_user",
        password="example-password"
    ) as connection:

        with connection.cursor() as cursor:
            # Remove money from the first account.
            cursor.execute(
                """
                UPDATE accounts
                SET balance = balance - %s
                WHERE id = %s
                """,
                (100.00, 1)
            )

            # Add money to the second account.
            cursor.execute(
                """
                UPDATE accounts
                SET balance = balance + %s
                WHERE id = %s
                """,
                (100.00, 2)
            )

    print("Transfer completed.")

except psycopg.Error as error:
    print("Transfer failed:", error)
Possible Output: 
Transfer completed.

Explanation: The connection context commits when the block finishes successfully. If a database exception occurs, the transaction is rolled back so that neither account is left partially updated.

41.11 Connection Pooling

Opening a new database connection for every request can be slow and use too many server resources. A connection pool keeps a limited collection of reusable connections. An application borrows a connection, performs its work, and returns the connection to the pool. Pool size should match application traffic and database-server capacity instead of being made unnecessarily large.

Example: PostgreSQL connection pool

# This example requires the Psycopg pool package.
# Install it with:
# python -m pip install psycopg_pool

from psycopg_pool import ConnectionPool

# Create a pool with a limited number of connections.
pool = ConnectionPool(
    conninfo=(
        "host=localhost "
        "dbname=store "
        "user=app_user "
        "password=example-password"
    ),
    min_size=1,
    max_size=5
)

# Borrow a connection from the pool.
with pool.connection() as connection:
    with connection.cursor() as cursor:
        cursor.execute("SELECT COUNT(*) FROM products")
        product_count = cursor.fetchone()[0]

        print("Products:", product_count)

# Close the pool when the application stops.
pool.close()
Example Output: 
Products: 24

Explanation: The application borrows one connection and returns it automatically. Later operations can reuse that connection instead of opening a completely new database session.

41.12 Prepared Statements

A prepared statement is an SQL statement that the database prepares before receiving its changing values. Prepared statements can improve safety and may improve performance when the same query runs repeatedly. Parameterized driver queries already separate values from SQL. Server-side prepared statements provide an additional database feature for repeated operations and controlled execution plans.

Example: PostgreSQL prepared statement

# Import the PostgreSQL driver.
import psycopg

with psycopg.connect(
    host="localhost",
    dbname="store",
    user="app_user",
    password="example-password"
) as connection:

    with connection.cursor() as cursor:
        # Prepare a statement on the PostgreSQL server.
        cursor.execute("""
            PREPARE find_product_by_id (INTEGER) AS
            SELECT id, name, price
            FROM products
            WHERE id = $1
        """)

        # Execute the prepared statement.
        cursor.execute(
            "EXECUTE find_product_by_id (%s)",
            (1,)
        )

        print(cursor.fetchone())

        # Remove the prepared statement.
        cursor.execute(
            "DEALLOCATE find_product_by_id"
        )
Example Output: 
(1, 'Keyboard', Decimal('49.99'))

Explanation: PostgreSQL prepares the query structure first and receives the product ID later. The final command removes the prepared statement from the current database session.

41.13 Database Relationships

Database relationships connect records stored in separate tables. A one-to-one relationship connects one record to one related record. A one-to-many relationship connects one parent to many children, such as one customer with many orders. A many-to-many relationship uses an intermediate table, such as students connected to many courses through an enrollments table.

Example: One-to-many relationship

CREATE TABLE customers (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name VARCHAR(100) NOT NULL
);

CREATE TABLE orders (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    total NUMERIC(10, 2) NOT NULL,
    FOREIGN KEY (customer_id)
        REFERENCES customers(id)
        ON DELETE RESTRICT
);

Example: Read related records

SELECT
    customers.name,
    orders.id,
    orders.total
FROM customers
INNER JOIN orders
    ON customers.id = orders.customer_id
ORDER BY customers.name;
Example Output: 
Sara | 101 | 45.00
Sara | 102 | 25.50
Ali  | 103 | 80.00

Explanation: One customer can appear beside several orders. The foreign key ensures that each order refers to an existing customer record.

41.14 Indexes

An index is a database structure that helps the server find rows faster. It works similarly to an index in a book. Indexes are useful for columns frequently used in searches, joins, sorting, and uniqueness checks. However, indexes require storage and can slow inserts, updates, and deletes because the index must also be maintained.

Example: Create an index

# Create an index for customer email searches.
CREATE INDEX idx_customers_email
ON customers(email);

# Create a multi-column index for order searches.
CREATE INDEX idx_orders_customer_date
ON orders(customer_id, created_at);

Example: Query that can use the index

SELECT id, name, email
FROM customers
WHERE email = 'sara@example.com';
Output: 
1 | Sara | sara@example.com

Explanation: The email index helps the server locate matching records without scanning every row. Index usefulness depends on the query, table size, and distribution of values.

41.15 Query Optimization

Query optimization means improving SQL so it uses less time, memory, processing power, and disk access. Developers should select only needed columns, filter early, use suitable indexes, avoid unnecessary repeated queries, and examine execution plans. Optimization should be based on measurements because a query that appears shorter is not always faster for the database.

Example: Avoid selecting unnecessary columns

# Less focused query:
SELECT *
FROM orders
WHERE customer_id = 10;

# More focused query:
SELECT id, total, created_at
FROM orders
WHERE customer_id = 10
ORDER BY created_at DESC
LIMIT 20;

Example: Examine a PostgreSQL query plan

EXPLAIN ANALYZE
SELECT id, total, created_at
FROM orders
WHERE customer_id = 10
ORDER BY created_at DESC
LIMIT 20;
Example Output: 
Limit
  -> Index Scan using idx_orders_customer_date on orders
Execution Time: 0.215 ms

Explanation: The execution plan shows that PostgreSQL used an index instead of reading the complete table. Actual output depends on the database contents, indexes, and server configuration.

41.16 Database Migrations

A database migration is a controlled change to the database structure. Migrations can create tables, add columns, build indexes, or transform existing data. Each migration should have a clear order and be tracked with the project code. Production databases should be backed up and tested before applying structural changes that could affect existing records.

Example: Add a phone column

ALTER TABLE customers
ADD COLUMN phone VARCHAR(30);

Example: Record a migration in Python

# Import the PostgreSQL driver.
import psycopg

migration_sql = """
    ALTER TABLE customers
    ADD COLUMN IF NOT EXISTS phone VARCHAR(30)
"""

with psycopg.connect(
    host="localhost",
    dbname="store",
    user="app_user",
    password="example-password"
) as connection:

    with connection.cursor() as cursor:
        cursor.execute(migration_sql)

print("Migration completed.")
Output: 
Migration completed.

Explanation: The migration adds a phone column. The IF NOT EXISTS clause prevents an error if the column already exists in a supported PostgreSQL setup.

41.17 Error Handling

Database operations can fail because of connection problems, invalid SQL, duplicate values, missing records, permission errors, timeouts, and broken constraints. Applications should catch expected database exceptions, roll back incomplete work, log useful technical details, and show users a safe message. Database passwords and full connection strings should never appear in error output.

Example: PostgreSQL error handling

# Import the PostgreSQL driver.
import psycopg
from psycopg.errors import UniqueViolation

try:
    with psycopg.connect(
        host="localhost",
        dbname="store",
        user="app_user",
        password="example-password"
    ) as connection:

        with connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO customers (name, email)
                VALUES (%s, %s)
                """,
                ("Sara", "sara@example.com")
            )

    print("Customer created.")

except UniqueViolation:
    print("That email address is already registered.")

except psycopg.OperationalError:
    print("The database server is unavailable.")

except psycopg.Error as error:
    print("A database error occurred.")
    print("Error type:", type(error).__name__)
Possible Output: 
That email address is already registered.

Explanation: The program handles a duplicate email separately from connection failures and other database errors. This produces clearer and safer application behavior.

41.18 Environment Configuration

Database settings should normally come from environment variables instead of being written directly in source code. This allows development, testing, and production systems to use different servers and credentials. The program should validate required settings before connecting. Secrets should be protected by the operating system, deployment platform, or a dedicated secret-management service.

Example: Load PostgreSQL settings

# Import the os module.
import os

# Load database settings.
database_config = {
    "host": os.getenv("DB_HOST", "localhost"),
    "port": int(os.getenv("DB_PORT", "5432")),
    "dbname": os.getenv("DB_NAME", "store"),
    "user": os.getenv("DB_USER"),
    "password": os.getenv("DB_PASSWORD")
}

# Validate required values.
if not database_config["user"]:
    raise ValueError("DB_USER is required.")

if not database_config["password"]:
    raise ValueError("DB_PASSWORD is required.")

# Display only safe settings.
print("Host:", database_config["host"])
print("Port:", database_config["port"])
print("Database:", database_config["dbname"])
print("User:", database_config["user"])
Example Environment Variables: 
DB_HOST=localhost
DB_PORT=5432
DB_NAME=store
DB_USER=app_user
DB_PASSWORD=protected-password
Output: 
Host: localhost
Port: 5432
Database: store
User: app_user

Explanation: The password is loaded but never printed. The program stops early with a clear error when a required username or password is missing.

41.19 Production Database Practices

Production databases require careful security, reliability, and monitoring. Applications should use limited database accounts, encrypted network connections, parameterized queries, connection pools, timeouts, backups, migration testing, and useful logs. Developers should monitor slow queries and connection usage. Database credentials must be rotated safely, and recovery procedures should be tested before an emergency happens.

Example: Safer PostgreSQL connection settings

# Import required modules.
import os
import psycopg

# Open a connection using protected environment settings.
connection = psycopg.connect(
    host=os.environ["DB_HOST"],
    port=int(os.getenv("DB_PORT", "5432")),
    dbname=os.environ["DB_NAME"],
    user=os.environ["DB_USER"],
    password=os.environ["DB_PASSWORD"],

    # Require an encrypted connection when supported.
    sslmode="require",

    # Stop waiting after a limited connection period.
    connect_timeout=10,

    # Give the connection an identifiable name.
    application_name="inventory_api"
)

print("Secure database connection created.")

connection.close()
Output: 
Secure database connection created.

Explanation: The connection requires encryption, uses a timeout, and identifies the application. The production account should receive only the permissions needed by the application.

41.20 Chapter Project

In this project, you will create a database-backed order-management application. The project uses PostgreSQL, but the overall structure can be adapted for MySQL. It stores customers, products, orders, and order items in related tables. It also uses environment variables, parameterized queries, transactions, connection pooling, indexes, validation, and database error handling.

Step 1: Create the project structure

order_manager/
│
├── app.py
├── config.py
├── database.py
├── schema.sql
└── order_service.py

Step 2: Create schema.sql

CREATE TABLE IF NOT EXISTS customers (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(255) NOT NULL UNIQUE,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS products (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name VARCHAR(150) NOT NULL,
    price NUMERIC(10, 2) NOT NULL CHECK (price >= 0),
    quantity INTEGER NOT NULL DEFAULT 0 CHECK (quantity >= 0),
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS orders (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    status VARCHAR(30) NOT NULL DEFAULT 'pending',
    total NUMERIC(10, 2) NOT NULL DEFAULT 0,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,

    FOREIGN KEY (customer_id)
        REFERENCES customers(id)
        ON DELETE RESTRICT
);

CREATE TABLE IF NOT EXISTS order_items (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    order_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    quantity INTEGER NOT NULL CHECK (quantity > 0),
    unit_price NUMERIC(10, 2) NOT NULL CHECK (unit_price >= 0),

    FOREIGN KEY (order_id)
        REFERENCES orders(id)
        ON DELETE CASCADE,

    FOREIGN KEY (product_id)
        REFERENCES products(id)
        ON DELETE RESTRICT
);

CREATE INDEX IF NOT EXISTS idx_customers_email
ON customers(email);

CREATE INDEX IF NOT EXISTS idx_orders_customer_id
ON orders(customer_id);

CREATE INDEX IF NOT EXISTS idx_order_items_order_id
ON order_items(order_id);

CREATE INDEX IF NOT EXISTS idx_order_items_product_id
ON order_items(product_id);

Step 3: Create config.py

# config.py

# Import the os module.
import os


# Create an application settings class.
class Settings:
    DB_HOST = os.getenv("DB_HOST", "localhost")
    DB_PORT = int(os.getenv("DB_PORT", "5432"))
    DB_NAME = os.getenv("DB_NAME", "order_manager")
    DB_USER = os.getenv("DB_USER")
    DB_PASSWORD = os.getenv("DB_PASSWORD")

    POOL_MIN_SIZE = int(
        os.getenv("POOL_MIN_SIZE", "1")
    )

    POOL_MAX_SIZE = int(
        os.getenv("POOL_MAX_SIZE", "5")
    )


# Validate all important settings.
def validate_settings():
    if not Settings.DB_USER:
        raise ValueError("DB_USER is required.")

    if not Settings.DB_PASSWORD:
        raise ValueError("DB_PASSWORD is required.")

    if not 1 <= Settings.DB_PORT <= 65535:
        raise ValueError("DB_PORT is invalid.")

    if Settings.POOL_MIN_SIZE < 1:
        raise ValueError(
            "POOL_MIN_SIZE must be at least 1."
        )

    if (
        Settings.POOL_MAX_SIZE
        < Settings.POOL_MIN_SIZE
    ):
        raise ValueError(
            "POOL_MAX_SIZE cannot be smaller "
            "than POOL_MIN_SIZE."
        )


# Build a PostgreSQL connection string.
def get_connection_string():
    return (
        f"host={Settings.DB_HOST} "
        f"port={Settings.DB_PORT} "
        f"dbname={Settings.DB_NAME} "
        f"user={Settings.DB_USER} "
        f"password={Settings.DB_PASSWORD} "
        f"connect_timeout=10 "
        f"application_name=order_manager"
    )

Step 4: Create database.py

# database.py

# Import Path for reading the schema file.
from pathlib import Path

# Import the PostgreSQL pool.
from psycopg_pool import ConnectionPool

# Import project configuration.
from config import (
    Settings,
    get_connection_string
)


# Create the reusable connection pool.
pool = ConnectionPool(
    conninfo=get_connection_string(),
    min_size=Settings.POOL_MIN_SIZE,
    max_size=Settings.POOL_MAX_SIZE,
    open=False
)


# Open the connection pool.
def open_pool():
    pool.open()
    pool.wait()


# Close the connection pool.
def close_pool():
    pool.close()


# Create all database tables and indexes.
def create_schema():
    # Read the SQL schema file.
    schema_sql = Path("schema.sql").read_text(
        encoding="utf-8"
    )

    # Borrow a connection from the pool.
    with pool.connection() as connection:
        # Execute the complete schema.
        with connection.cursor() as cursor:
            cursor.execute(schema_sql)

Step 5: Create order_service.py

# order_service.py

# Import decimal values for money.
from decimal import Decimal

# Import the connection pool.
from database import pool


# Add a customer.
def add_customer(name, email):
    # Validate the supplied values.
    if not name.strip():
        raise ValueError(
            "Customer name cannot be empty."
        )

    if not email.strip():
        raise ValueError(
            "Customer email cannot be empty."
        )

    # Borrow a database connection.
    with pool.connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO customers (name, email)
                VALUES (%s, %s)
                RETURNING id
                """,
                (name.strip(), email.strip())
            )

            return cursor.fetchone()[0]


# Add a product.
def add_product(name, price, quantity):
    # Convert and validate the price.
    price = Decimal(str(price))

    if price < 0:
        raise ValueError(
            "Product price cannot be negative."
        )

    # Validate the quantity.
    if quantity < 0:
        raise ValueError(
            "Product quantity cannot be negative."
        )

    # Insert the product.
    with pool.connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO products (
                    name,
                    price,
                    quantity
                )
                VALUES (%s, %s, %s)
                RETURNING id
                """,
                (
                    name.strip(),
                    price,
                    quantity
                )
            )

            return cursor.fetchone()[0]


# Return all available products.
def get_products():
    with pool.connection() as connection:
        with connection.cursor() as cursor:
            cursor.execute("""
                SELECT id, name, price, quantity
                FROM products
                ORDER BY name
            """)

            return cursor.fetchall()


# Create an order containing several items.
def create_order(customer_id, items):
    # Items should look like:
    # [
    #     {"product_id": 1, "quantity": 2},
    #     {"product_id": 2, "quantity": 1}
    # ]

    if not items:
        raise ValueError(
            "An order must contain at least one item."
        )

    # Borrow one connection for the transaction.
    with pool.connection() as connection:
        with connection.cursor() as cursor:
            # Create the empty order.
            cursor.execute(
                """
                INSERT INTO orders (
                    customer_id,
                    status,
                    total
                )
                VALUES (%s, %s, %s)
                RETURNING id
                """,
                (
                    customer_id,
                    "pending",
                    Decimal("0.00")
                )
            )

            order_id = cursor.fetchone()[0]
            order_total = Decimal("0.00")

            # Process every requested item.
            for item in items:
                product_id = item["product_id"]
                requested_quantity = item["quantity"]

                if requested_quantity <= 0:
                    raise ValueError(
                        "Item quantity must be positive."
                    )

                # Lock the product row during checkout.
                cursor.execute(
                    """
                    SELECT name, price, quantity
                    FROM products
                    WHERE id = %s
                    FOR UPDATE
                    """,
                    (product_id,)
                )

                product = cursor.fetchone()

                if product is None:
                    raise ValueError(
                        f"Product {product_id} was not found."
                    )

                product_name = product[0]
                product_price = product[1]
                available_quantity = product[2]

                if available_quantity < requested_quantity:
                    raise ValueError(
                        f"Not enough inventory for "
                        f"{product_name}."
                    )

                # Add the order item.
                cursor.execute(
                    """
                    INSERT INTO order_items (
                        order_id,
                        product_id,
                        quantity,
                        unit_price
                    )
                    VALUES (%s, %s, %s, %s)
                    """,
                    (
                        order_id,
                        product_id,
                        requested_quantity,
                        product_price
                    )
                )

                # Reduce the available inventory.
                cursor.execute(
                    """
                    UPDATE products
                    SET quantity = quantity - %s
                    WHERE id = %s
                    """,
                    (
                        requested_quantity,
                        product_id
                    )
                )

                # Add the item value to the order total.
                order_total += (
                    product_price
                    * requested_quantity
                )

            # Store the completed total.
            cursor.execute(
                """
                UPDATE orders
                SET total = %s,
                    status = %s
                WHERE id = %s
                """,
                (
                    order_total,
                    "confirmed",
                    order_id
                )
            )

            # Return information about the new order.
            return order_id, order_total


# Return complete order details.
def get_order(order_id):
    with pool.connection() as connection:
        with connection.cursor() as cursor:
            # Read the main order information.
            cursor.execute(
                """
                SELECT
                    orders.id,
                    customers.name,
                    customers.email,
                    orders.status,
                    orders.total,
                    orders.created_at
                FROM orders
                INNER JOIN customers
                    ON orders.customer_id = customers.id
                WHERE orders.id = %s
                """,
                (order_id,)
            )

            order = cursor.fetchone()

            if order is None:
                return None

            # Read every order item.
            cursor.execute(
                """
                SELECT
                    products.name,
                    order_items.quantity,
                    order_items.unit_price,
                    (
                        order_items.quantity
                        * order_items.unit_price
                    ) AS line_total
                FROM order_items
                INNER JOIN products
                    ON order_items.product_id = products.id
                WHERE order_items.order_id = %s
                ORDER BY products.name
                """,
                (order_id,)
            )

            items = cursor.fetchall()

            return {
                "order": order,
                "items": items
            }

Step 6: Create app.py

# app.py

# Import PostgreSQL exceptions.
import psycopg
from psycopg.errors import UniqueViolation

# Import project configuration.
from config import validate_settings

# Import database functions.
from database import (
    open_pool,
    close_pool,
    create_schema
)

# Import order operations.
from order_service import (
    add_customer,
    add_product,
    get_products,
    create_order,
    get_order
)


# Display the available menu.
def show_menu():
    print("\nOrder Manager")
    print("1. Add customer")
    print("2. Add product")
    print("3. List products")
    print("4. Create order")
    print("5. View order")
    print("6. Exit")


# Display all products.
def display_products():
    products = get_products()

    if not products:
        print("No products were found.")
        return

    for product in products:
        print(
            f"{product[0]}: {product[1]} | "
            f"${product[2]} | "
            f"Quantity: {product[3]}"
        )


# Run the application.
def main():
    # Validate settings before opening connections.
    validate_settings()

    # Open the shared connection pool.
    open_pool()

    try:
        # Create all tables and indexes.
        create_schema()

        while True:
            show_menu()

            choice = input(
                "Choose an option: "
            ).strip()

            try:
                if choice == "1":
                    name = input(
                        "Customer name: "
                    ).strip()

                    email = input(
                        "Customer email: "
                    ).strip()

                    customer_id = add_customer(
                        name,
                        email
                    )

                    print(
                        f"Customer created with "
                        f"ID {customer_id}."
                    )

                elif choice == "2":
                    name = input(
                        "Product name: "
                    ).strip()

                    price = input(
                        "Product price: "
                    ).strip()

                    quantity = int(
                        input("Product quantity: ")
                    )

                    product_id = add_product(
                        name,
                        price,
                        quantity
                    )

                    print(
                        f"Product created with "
                        f"ID {product_id}."
                    )

                elif choice == "3":
                    display_products()

                elif choice == "4":
                    customer_id = int(
                        input("Customer ID: ")
                    )

                    item_count = int(
                        input(
                            "Number of different items: "
                        )
                    )

                    items = []

                    for number in range(
                        1,
                        item_count + 1
                    ):
                        print(f"Item {number}")

                        product_id = int(
                            input("Product ID: ")
                        )

                        quantity = int(
                            input("Quantity: ")
                        )

                        items.append({
                            "product_id": product_id,
                            "quantity": quantity
                        })

                    order_id, total = create_order(
                        customer_id,
                        items
                    )

                    print(
                        f"Order {order_id} created."
                    )

                    print(
                        f"Order total: ${total}"
                    )

                elif choice == "5":
                    order_id = int(
                        input("Order ID: ")
                    )

                    result = get_order(order_id)

                    if result is None:
                        print("Order not found.")
                        continue

                    order = result["order"]

                    print("\nOrder Information")
                    print("Order ID:", order[0])
                    print("Customer:", order[1])
                    print("Email:", order[2])
                    print("Status:", order[3])
                    print("Total:", order[4])
                    print("Created:", order[5])

                    print("\nOrder Items")

                    for item in result["items"]:
                        print(
                            f"{item[0]} | "
                            f"Quantity: {item[1]} | "
                            f"Unit price: ${item[2]} | "
                            f"Line total: ${item[3]}"
                        )

                elif choice == "6":
                    print("Goodbye.")
                    break

                else:
                    print("Invalid option.")

            except ValueError as error:
                print("Input error:", error)

            except UniqueViolation:
                print(
                    "A record with that unique "
                    "value already exists."
                )

            except psycopg.OperationalError:
                print(
                    "The database connection was lost."
                )

            except psycopg.Error as error:
                print(
                    "A database operation failed."
                )

                print(
                    "Error type:",
                    type(error).__name__
                )

    finally:
        # Always close the connection pool.
        close_pool()


# Start the application.
if __name__ == "__main__":
    main()

Step 7: Set environment variables

DB_HOST=localhost
DB_PORT=5432
DB_NAME=order_manager
DB_USER=app_user
DB_PASSWORD=protected-password
POOL_MIN_SIZE=1
POOL_MAX_SIZE=5

Step 8: Install the required packages

python -m pip install psycopg[binary] psycopg_pool

Step 9: Run the project

python app.py
Example Output: 
Order Manager
1. Add customer
2. Add product
3. List products
4. Create order
5. View order
6. Exit

Choose an option: 1
Customer name: Sara Smith
Customer email: sara@example.com
Customer created with ID 1.

Choose an option: 2
Product name: Keyboard
Product price: 49.99
Product quantity: 10
Product created with ID 1.

Choose an option: 2
Product name: Mouse
Product price: 24.50
Product quantity: 20
Product created with ID 2.

Choose an option: 4
Customer ID: 1
Number of different items: 2

Item 1
Product ID: 1
Quantity: 2

Item 2
Product ID: 2
Quantity: 1

Order 1 created.
Order total: $124.48

Choose an option: 5
Order ID: 1

Order Information
Order ID: 1
Customer: Sara Smith
Email: sara@example.com
Status: confirmed
Total: 124.48
Created: 2026-07-19 18:30:15

Order Items
Keyboard | Quantity: 2 | Unit price: $49.99 | Line total: $99.98
Mouse | Quantity: 1 | Unit price: $24.50 | Line total: $24.50

Explanation: This project uses PostgreSQL tables, primary keys, foreign keys, indexes, parameterized queries, transactions, row locking, and connection pooling. Creating an order is one transaction. If a product is missing or inventory is insufficient, the complete order operation is rolled back, preventing incomplete orders and incorrect stock values.

End of Chapter 41: PostgreSQL and MySQL

Chapter 2: Development EnvironmentPrepare Your JavaScript Workspace

A modern course built to help learners study step by step with clarity, comfort, and confidence.