MySQL & PostgreSQL Database Drivers in Python

While SQLite is excellent for embedded local storage and testing, enterprise Python services typically run against client-server database engines such as MySQL and PostgreSQL. In this session, you will learn how to interface with production databases using standard third-party driver libraries.

1. Driver Comparison & Installation

Unlike sqlite3, external drivers must be installed into your virtual environment via pip:

# MySQL official driver pip install mysql-connector-python # PostgreSQL driver (C-extension or binary distribution) pip install psycopg2-binary

Feature	MySQL (`mysql-connector-python`)	PostgreSQL (`psycopg2`)
Default Port	`3306`	`5432`
Placeholder Style	`%s`	`%s`
Dictionary Cursor	`cursor(dictionary=True)`	`psycopg2.extras.RealDictCursor`
Connection Pool	`mysql.connector.pooling`	`psycopg2.pool`

2. Connecting to MySQL in Python

This example connects to a MySQL server, executes a parameterized query, and uses a dictionary cursor to return rows as key-value pairs.

import os import mysql.connector # Fetch credentials from environment variables db_config = { 'host': os.getenv('DB_HOST', 'localhost'), 'user': os.getenv('DB_USER', 'app_user'), 'password': os.getenv('DB_PASS', 'secure_password'), 'database': os.getenv('DB_NAME', 'production_db'), 'port': 3306 } try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor(dictionary=True) cursor.execute("SELECT id, username, email FROM users WHERE active = %s", (1,)) rows = cursor.fetchall() for row in rows: print(f"User: {row['username']} | Email: {row['email']}") except mysql.connector.Error as err: print(f"MySQL Error: {err}") finally: if 'conn' in locals() and conn.is_connected(): cursor.close() conn.close()

3. Connecting to PostgreSQL in Python

PostgreSQL uses psycopg2. Using RealDictCursor allows returned tuples to behave like standard Python dictionaries.

import os import psycopg2 from psycopg2.extras import RealDictCursor try: conn = psycopg2.connect( host=os.getenv('PG_HOST', 'localhost'), database=os.getenv('PG_DB', 'analytics'), user=os.getenv('PG_USER', 'pg_admin'), password=os.getenv('PG_PASS', 'secret'), port=5432 ) # Enable dict results with conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute("SELECT * FROM metrics WHERE timestamp >= NOW() - INTERVAL '1 hour'") results = cursor.fetchall() print("Fetched records:", len(results)) conn.commit() except psycopg2.DatabaseError as err: print(f"PostgreSQL Error: {err}") finally: if 'conn' in locals() and conn: conn.close()

4. Production Connection Pooling Pattern

Opening and closing socket connections for every API request creates high network latency. Connection pools maintain reusable open sockets to improve performance in multi-threaded applications.

# Connection Pooling Example using MySQL Connector from mysql.connector import pooling pool = pooling.MySQLConnectionPool( pool_name="web_app_pool", pool_size=5, host="localhost", user="root", password="password", database="app_db" ) # Request connection from pool connection = pool.get_connection() cursor = connection.cursor() cursor.execute("SELECT 1") print("Pool Query Success:", cursor.fetchone()) # Returning connection back to pool (does not close socket) cursor.close() connection.close()

Production Best Practice: Never hardcode database passwords in your codebase. Always use environment variables or secret vaults (e.g., AWS Secrets Manager, HashiCorp Vault).

🏋️ Try It Yourself: Practice Challenges

Challenge 1: Environment Configuration Builder

Write a Python function that builds a database connection dictionary from environment variables, providing default fallbacks for local development.

Challenge 2: Query Abstraction Interface

Create a helper class DatabaseManager that wraps query execution in try...except...finally blocks to handle connection cleanup automatically.

Challenge 3: Connection Pool Wrapper

Implement a context manager function that retrieves a connection from a pool and returns it automatically upon completion.

⚡ Interactive Driver Emulator Sandbox

# Simulation of Enterprise DB Adapter Pattern using SQLite Engine
import sqlite3
import os

class ProductionDBAdapter:
    def __init__(self, db_type="mysql"):
        self.db_type = db_type
        # Emulating external DB with memory storage
        self.conn = sqlite3.connect(':memory:')
        self.conn.row_factory = sqlite3.Row
        self._bootstrap_schema()

    def _bootstrap_schema(self):
        cur = self.conn.cursor()
        cur.execute('''
            CREATE TABLE logs (
                id INTEGER PRIMARY KEY,
                service TEXT NOT NULL,
                message TEXT NOT NULL
            )
        ''')
        cur.execute("INSERT INTO logs (service, message) VALUES ('auth_service', 'User Login Successful')")
        self.conn.commit()

    def fetch_logs(self, service_name):
        # Universal placeholder pattern
        query = "SELECT * FROM logs WHERE service = ?"
        cur = self.conn.cursor()
        cur.execute(query, (service_name,))
        return [dict(row) for row in cur.fetchall()]

# Demonstration
db = ProductionDBAdapter(db_type="postgresql")
logs = db.fetch_logs("auth_service")

print(f"Emulated DB [{db.db_type.upper()}] Query Result:")
print(logs)

Console Output:

Click "Run Code" above to execute driver emulation script...

📝 Knowledge Check Quiz

1. What is the main advantage of using connection pooling in production applications?

It automatically encrypts all database tables It reuses open database connections to reduce handshake latency It converts SQL queries to Python functions

2. Which PostgreSQL driver library is widely used in standard Python enterprise development?

sqlite3 psycopg2 py-postgres-connect

3. What is the recommended way to handle database passwords in production Python applications?

Store credentials in environment variables or secret management services Write passwords directly into source files Pass raw passwords as URL parameters

Horizontal Ad Banner Slot (728x90 / Responsive)