Database Connections

🗄️ Database Connections

💡 Why Database Connections Matter

A database connection is like a phone call between your Python script and the database server. You need to:

1. Establish the connection — Dial the number and wait for pickup
2. Send queries — Ask questions and get answers
3. Handle errors — What if the database is unavailable?
4. Close the connection — Hang up when done

Many bugs happen because developers forget step 4 — connections that never close consume server resources and eventually crash the server.

📦 Installing mysql-connector-python

Copypip install mysql-connector-python==8.2.0

Add to requirements.txt:

Copymysql-connector-python==8.2.0
python-dotenv==1.0.0

🔌 Connection Basics

The Simplest Connection

Copyimport mysql.connector

# Create a connection
connection = mysql.connector.connect(
    host="localhost",
    user="philip",
    password="AsT@1sAd3mon",
    database="learning_blog"
)

print("✅ Connected to database!")

# When done, close it
connection.close()
print("✅ Connection closed")

Connection Parameters

↔️ Cursors: Executing Queries

Creating and Using a Cursor

Copyimport mysql.connector

connection = mysql.connector.connect(
    host="localhost",
    user="philip",
    password="AsT@1sAd3mon",
    database="learning_blog"
)

# Create a cursor from the connection
cursor = connection.cursor()

# Execute a query
cursor.execute("SELECT COUNT(*) FROM pages")

# Fetch the result
result = cursor.fetchone()
count = result[0]

print(f"Total pages: {count}")

# Clean up
cursor.close()
connection.close()

Fetch Methods

Copy# SELECT COUNT(*) FROM pages  → 1 row with 1 column
result = cursor.fetchone()
# Returns: (42,)  ← tuple with one element

# SELECT id, title FROM pages  → many rows
rows = cursor.fetchall()
# Returns: [(1, 'Chapter 1'), (2, 'Chapter 2'), ...]

# Iterate over results
for row in rows:
    id, title = row
    print(f"{id}: {title}")

# Get column count
num_columns = cursor.rowcount
print(f"Affected {num_columns} rows")

⚠️ Error Handling

Catching Connection Errors

Copyimport mysql.connector
from mysql.connector import Error

try:
    connection = mysql.connector.connect(
        host="localhost",
        user="philip",
        password="wrong_password",
        database="learning_blog"
    )

except Error as e:
    print(f"❌ Connection failed: {e}")
    print(f"   Error code: {e.errno}")
    print(f"   Error message: {e.msg}")

else:
    print("✅ Connected successfully!")
    connection.close()

Common Error Codes

🎯 Context Managers: The Right Way

The Wrong Way (Connections Can Leak)

Copy# ❌ RISKY: If an exception occurs, close() never runs
connection = mysql.connector.connect(...)
cursor = connection.cursor()
cursor.execute("SELECT * FROM pages")
connection.close()  # ← Never reached if error occurs!

The Right Way (With Context Manager)

Copyimport mysql.connector
from contextlib import closing

connection = mysql.connector.connect(
    host="localhost",
    user="philip",
    password="AsT@1sAd3mon",
    database="learning_blog"
)

# ✅ SAFE: closing() ensures close() is always called
with closing(connection.cursor()) as cursor:
    cursor.execute("SELECT COUNT(*) FROM pages")
    result = cursor.fetchone()
    print(f"Pages: {result[0]}")

# At this point, cursor is closed automatically!
connection.close()

🌍 Real-World Example: DatabaseConnection Class

Here's how db_utils.py structures database operations:

Copyclass DatabaseConnection:
    def __init__(self, host, user, password, database, port=3306):
        self.config = {
            'host': host,
            'user': user,
            'password': password,
            'database': database,
            'port': port,
            'charset': 'utf8mb4'
        }
        self.connection = None

    def connect(self) -> bool:
        """Establish database connection."""
        try:
            self.connection = mysql.connector.connect(**self.config)
            return True
        except Error as e:
            print(f"❌ Connection failed: {e}")
            return False

    def execute_query(self, query: str) -> list:
        """Execute a SELECT query and return results."""
        try:
            cursor = self.connection.cursor()
            cursor.execute(query)
            results = cursor.fetchall()
            cursor.close()
            return results
        except Error as e:
            print(f"❌ Query failed: {e}")
            return []

    def disconnect(self):
        """Close the database connection."""
        if self.connection:
            self.connection.close()

# Usage:
db = DatabaseConnection("localhost", "philip", "password", "learning_blog")
if db.connect():
    results = db.execute_query("SELECT * FROM pages")
    db.disconnect()

⚠️ Common Mistakes

Copyconnection = mysql.connector.connect(...)
cursor = connection.cursor()
cursor.execute("SELECT ...")
# ❌ Never closed! Connection leaks!

Copytry:
    connection = mysql.connector.connect(...)
    cursor = connection.cursor()
    cursor.execute("SELECT ...")
finally:
    cursor.close()
    connection.close()

Copyconnection = mysql.connector.connect(host="wrong_host", ...)
# ❌ If connection fails, this crashes!

Copytry:
    connection = mysql.connector.connect(...)
except Error as e:
    print(f"Connection failed: {e}")
    sys.exit(1)

Copy# SELECT id, title, created_at FROM pages
row = cursor.fetchone()
print(row[5])  # ❌ IndexError! Only 3 columns (0, 1, 2)

Copyid, title, created = row  # ✅ Unpack correctly
print(id, title, created)

💻 Coding Challenges

🎯 What's Next

You now have the tools to connect to databases. Chapter 4 covers Building db_utils.py — a complete breakdown of the actual database utility script used in production, line-by-line explanation of every function.