Python Database Access

Python Database Access (using MySQL Connector)

Python provides a standardized way to interact with various databases through the Python Database API Specification v2.0 (DB-API 2). Most database drivers/connectors for Python adhere to this standard.

You need to install a specific DB-API compliant driver for the database you want to connect to (e.g., PostgreSQL, SQLite, Oracle, SQL Server, MySQL). This tutorial focuses on MySQL using the official mysql-connector-python driver.

The DB-API workflow generally involves:

Installing mysql-connector-python

Before you start, install the driver using pip:

pip install mysql-connector-python

Verify the installation by trying to import it in a Python interpreter:

import mysql.connector

If this runs without an ImportError, the driver is installed.

Prerequisites

For the examples below, assume:

Establishing a Database Connection

Use mysql.connector.connect() to establish a connection. It returns a connection object.

import mysql.connector
from mysql.connector import Error

def create_connection(host_name, user_name, user_password, db_name):
    """Create a database connection to the MySQL database specified."""
    connection = None
    try:
        connection = mysql.connector.connect(
            host=host_name,
            user=user_name,
            password=user_password,
            database=db_name
        )
        print("MySQL Database connection successful")
    except Error as e:
        print(f"The error '{e}' occurred")
    return connection

# Example usage (replace with your details)
conn = create_connection("localhost", "testuser", "test1234", "testdb")

# Remember to close the connection when done
# if conn and conn.is_connected():
#    conn.close()
#    print("MySQL connection is closed")

Using Context Managers (with statement) is the recommended way to handle connections and cursors, as they ensure resources are closed automatically.

Creating a Database Table

Once connected, create a cursor object using connection.cursor(). Use the cursor’s execute() method to run SQL DDL (Data Definition Language) commands like CREATE TABLE.

import mysql.connector
from mysql.connector import Error

# Assume create_connection function exists as defined above

def execute_query(connection, query):
    """Executes a single SQL query."""
    cursor = connection.cursor()
    try:
        cursor.execute(query)
        connection.commit() # Important for DDL/DML if autocommit=False
        print("Query executed successfully")
    except Error as e:
        print(f"The error '{e}' occurred")
    finally:
        if cursor:
             cursor.close()

# -- Main execution --
conn = create_connection("localhost", "testuser", "test1234", "testdb")

if conn and conn.is_connected():
    # Drop table if it exists (for demonstration)
    drop_table_query = "DROP TABLE IF EXISTS employees;"
    execute_query(conn, drop_table_query)

    # Create the employees table
    create_employees_table = """
    CREATE TABLE employees (
      id INT AUTO_INCREMENT PRIMARY KEY,
      first_name VARCHAR(50) NOT NULL,
      last_name VARCHAR(50),
      age INT,
      gender CHAR(1),
      salary FLOAT
    );
    """
    execute_query(conn, create_employees_table)

    conn.close()
    print("MySQL connection is closed")

INSERT Operation (Creating Records)

Use the execute() method with an INSERT statement. Crucially, use parameterized queries (placeholders like %s) to prevent SQL injection vulnerabilities. Pass the values as a tuple or list as the second argument to execute().

# (Continuing from previous example, assuming connection 'conn' is open)

def insert_employee(connection, employee_data):
    """Inserts a new employee into the employees table."""
    sql = """
    INSERT INTO employees (first_name, last_name, age, gender, salary)
    VALUES (%s, %s, %s, %s, %s);
    """
    cursor = connection.cursor()
    try:
        cursor.execute(sql, employee_data)
        connection.commit() # Commit the transaction
        print(f"Employee {employee_data[0]} added successfully. ID: {cursor.lastrowid}")
        return cursor.lastrowid
    except Error as e:
        print(f"Failed to insert record: {e}")
        connection.rollback() # Rollback on error
    finally:
        if cursor:
            cursor.close()
    return None

# -- Inside main execution block --
if conn and conn.is_connected():
    # ... (table creation code) ...

    # Insert a single employee
    employee1 = ('Maria', 'Garcia', 30, 'F', 55000.00)
    insert_employee(conn, employee1)

    # Insert multiple employees using executemany
    employees_to_add = [
        ('John', 'Doe', 45, 'M', 75000.00),
        ('Lisa', 'Ray', 28, 'F', 62000.00)
    ]
    sql_insert_many = "INSERT INTO employees (first_name, last_name, age, gender, salary) VALUES (%s, %s, %s, %s, %s);"
    cursor = conn.cursor()
    try:
        cursor.executemany(sql_insert_many, employees_to_add)
        conn.commit()
        print(f"{cursor.rowcount} employees inserted successfully.")
    except Error as e:
        print(f"Failed to insert multiple records: {e}")
        conn.rollback()
    finally:
        if cursor:
            cursor.close()

    # ... (connection close) ...

READ Operation (Fetching Data)

Use execute() with a SELECT statement. Then use fetch methods like fetchone(), fetchall(), or fetchmany(size) on the cursor to retrieve the results.

# (Continuing from previous example, assuming connection 'conn' is open)

def fetch_all_employees(connection):
    """Fetches all employee records."""
    sql = "SELECT id, first_name, last_name, age, salary FROM employees;"
    cursor = connection.cursor(dictionary=True) # Fetch as dictionaries
    employees = []
    try:
        cursor.execute(sql)
        employees = cursor.fetchall()
        print(f"Fetched {len(employees)} employee records.")
    except Error as e:
        print(f"Failed to fetch records: {e}")
    finally:
        if cursor:
            cursor.close()
    return employees

# -- Inside main execution block --
if conn and conn.is_connected():
    # ... (table creation, insertion code) ...

    all_employees = fetch_all_employees(conn)
    if all_employees:
        for employee in all_employees:
             # Access by column name if dictionary=True, else by index (e.g., employee[1])
            print(f"ID: {employee['id']}, Name: {employee['first_name']} {employee['last_name']}, Age: {employee['age']}")

    # ... (connection close) ...

Setting dictionary=True in connection.cursor(dictionary=True) makes fetch methods return rows as dictionaries (column name -> value) instead of tuples, which is often more convenient.

UPDATE Operation

Use an UPDATE statement with execute(). Remember to use parameterized queries and commit the transaction.

# (Continuing, assuming 'conn' is open)

def update_employee_salary(connection, employee_id, new_salary):
    """Updates the salary for a given employee ID."""
    sql = "UPDATE employees SET salary = %s WHERE id = %s;"
    cursor = connection.cursor()
    try:
        cursor.execute(sql, (new_salary, employee_id))
        connection.commit()
        if cursor.rowcount > 0:
             print(f"Salary updated for employee ID {employee_id}.")
        else:
             print(f"No employee found with ID {employee_id}.")
    except Error as e:
        print(f"Failed to update record: {e}")
        connection.rollback()
    finally:
        if cursor:
            cursor.close()

# -- Inside main execution block --
if conn and conn.is_connected():
    # ... (creation, insertion, fetch code) ...
    update_employee_salary(conn, 1, 60000.00) # Update salary for employee with ID 1
    # ... (connection close) ...

DELETE Operation

Use a DELETE statement with execute(). Use parameterized queries and commit.

# (Continuing, assuming 'conn' is open)

def delete_employee(connection, employee_id):
    """Deletes an employee with the given ID."""
    sql = "DELETE FROM employees WHERE id = %s;"
    cursor = connection.cursor()
    try:
        cursor.execute(sql, (employee_id,))
        connection.commit()
        if cursor.rowcount > 0:
             print(f"Employee with ID {employee_id} deleted.")
        else:
             print(f"No employee found with ID {employee_id}.")
    except Error as e:
        print(f"Failed to delete record: {e}")
        connection.rollback()
    finally:
        if cursor:
            cursor.close()

# -- Inside main execution block --
if conn and conn.is_connected():
    # ... (creation, insertion, fetch, update code) ...
    delete_employee(conn, 3) # Delete employee with ID 3
    # ... (connection close) ...

Transactions (commit() and rollback())

Database transactions ensure data integrity using the ACID properties (Atomicity, Consistency, Isolation, Durability). Operations like INSERT, UPDATE, DELETE are usually part of a transaction.

It’s crucial to commit changes after successful DML operations or rollback in case of errors, as shown in the INSERT, UPDATE, and DELETE examples. By default, mysql-connector-python might have autocommit turned off, requiring explicit commits.

Disconnecting from the Database (close())

Always close the database connection when you are finished to release resources.

if conn and conn.is_connected():
    conn.close()
    print("MySQL connection is closed")

Using a with statement for the connection and cursor handles closing automatically, even if errors occur.

# Recommended pattern using 'with'
try:
    with create_connection("localhost", "testuser", "test1234", "testdb") as conn:
        if conn and conn.is_connected():
            with conn.cursor(dictionary=True) as cursor:
                cursor.execute("SELECT * FROM employees WHERE age > %s", (30,))
                results = cursor.fetchall()
                for row in results:
                    print(row)
            # Cursor is automatically closed here
        # Connection is automatically closed here (or rolled back on error within 'with')
        # Note: commit might still be needed explicitly depending on autocommit settings
except Error as e:
    print(f"Database error: {e}")
except Exception as e:
    print(f"An error occurred: {e}")

Error Handling

Database operations can fail for various reasons (invalid SQL, connection issues, constraint violations). Use try...except blocks to catch errors. The mysql-connector module raises exceptions derived from mysql.connector.Error.

Catch specific errors or the base Error class as needed.