Database Connections

Practical Python Scripting — Database Connections

Practical Python Scripting

Course 1 · Chapter 3 · Database Connections

🗄️ Database Connections

Now that you can load configuration securely, you're ready to connect to a database. This chapter teaches how to use the mysql-connector-python library to connect to MySQL/MariaDB, execute queries, handle errors, and clean up resources properly. You'll learn the patterns used in production code.

💡 Why Database Connections Matter

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

Establish the connection — Dial the number and wait for pickup
Send queries — Ask questions and get answers
Handle errors — What if the database is unavailable?
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

pip install mysql-connector-python==8.2.0

Add to requirements.txt:

mysql-connector-python==8.2.0
python-dotenv==1.0.0
        

✅ WHAT IT IS: mysql-connector-python is a pure Python implementation of the MySQL protocol. It doesn't require any C libraries, so it works on any system.

🔌 Connection Basics

The Simplest Connection

import 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")
        

✅ WHAT IT DOES: mysql.connector.connect() creates a connection to the database server. You provide credentials and database name.

Connection Parameters

Parameter	Required	Example	Description
`host`	Yes	`"localhost"`	Database server address
`user`	Yes	`"philip"`	Database username
`password`	Yes	`"secret123"`	Database password
`database`	Yes	`"learning_blog"`	Database name
`port`	No	`3306`	Port (default: 3306)
`charset`	No	`"utf8mb4"`	Character set (default: utf8mb4)
`autocommit`	No	`True`	Auto-commit changes

↔️ Cursors: Executing Queries

A cursor is how you send SQL queries to the database. You create a cursor from a connection, then use it to execute queries and fetch results.

Creating and Using a Cursor

import 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()
        

✅ WHAT IT DOES:

cursor() — Create a cursor from a connection
execute(sql) — Send a SQL query
fetchone() — Get one row of results
fetchall() — Get all rows of results

Fetch Methods

# 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

Database operations fail for many reasons: server offline, wrong credentials, permission denied, network timeout. Always use try/except to handle these gracefully.

Catching Connection Errors

import 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

Error Code	Meaning	Solution
1045	Access denied (wrong password/user)	Check credentials in .env
1049	Unknown database	Check database name
2003	Can't connect to server	Check if server is running
2006	MySQL server has gone away	Connection timed out, reconnect

🎯 Context Managers: The Right Way

Using try/except with close() is error-prone. If an exception occurs before close(), the connection stays open forever. The with statement (context manager) handles this automatically.

The Wrong Way (Connections Can Leak)

# ❌ 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)

import 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()
        

🤔 WHY CONTEXT MANAGERS: The with statement guarantees cleanup code runs, even if exceptions occur. This prevents resource leaks that crash servers.

🌍 Real-World Example: DatabaseConnection Class

Here's how db_utils.py structures database operations:

class 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

❌ MISTAKE 1: Forgetting to close connections

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

Fix:

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

❌ MISTAKE 2: Not handling connection errors

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

Fix:

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

❌ MISTAKE 3: Accessing columns by wrong index

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

Fix:

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

💻 Coding Challenges

Challenge 1: Test Database Connection

Create a script that:

Loads configuration from .env
Attempts to connect to the database
Shows success/failure message with error details
Safely closes the connection

Goal: Practice basic connection and error handling.

→ Solution

Challenge 2: Query and Display Results

Create a script that:

Connects to the database
Executes a SELECT query
Displays results in a formatted table
Handles empty results gracefully

Goal: Practice executing queries and displaying results.

→ Solution

Challenge 3: DatabaseConnection Class

Create a reusable DatabaseConnection class that:

Accepts configuration parameters
Has connect/disconnect methods
Has execute_query method for SELECT
Has execute_update method for INSERT/UPDATE/DELETE
Handles all errors gracefully

Goal: Build a reusable database utility class.

→ Solution

🎯 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.