How to Wrap MySQL Operations in Pytest for Seamless API Testing

1. Encapsulating MySQL Database Connection

Install the pymysql library, which allows Python to connect to MySQL. pip install pymysql Create a database.py file and define a MySQLDatabase class that stores connection parameters and provides methods for connecting, executing queries, and closing the connection.

import pymysql
import json
from loguru import logger

class MySQLDatabase:
    def __init__(self, host, user, password, db):
        self.host = host
        self.user = user
        self.password = password
        self.db = db
        self.connection = None

    def connect(self):
        """Establish a database connection"""
        try:
            self.connection = pymysql.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                db=self.db,
                charset='utf8mb4',
                cursorclass=pymysql.cursors.DictCursor
            )
            logger.info("Database connection successful")
        except Exception as e:
            logger.error(f"Database connection failed: {e}")
            raise

    def execute_query(self, query, params=None):
        """Execute a SELECT query and return JSON data"""
        if not self.connection:
            self.connect()
        try:
            with self.connection.cursor() as cursor:
                cursor.execute(query, params)
                result = cursor.fetchall()
                return json.dumps(result, ensure_ascii=False)  # JSON string
        except Exception as e:
            logger.error(f"Query execution failed: {e}")
            raise

    def execute_update(self, query, params=None):
        """Execute INSERT/UPDATE/DELETE and return affected rows"""
        if not self.connection:
            self.connect()
        try:
            with self.connection.cursor() as cursor:
                cursor.execute(query, params)
                self.connection.commit()
                return cursor.rowcount
        except Exception as e:
            self.connection.rollback()
            logger.error(f"Update execution failed: {e}")
            raise

    def close(self):
        """Close the database connection"""
        if self.connection:
            self.connection.close()
            logger.info("Database connection closed")

2. Encapsulating CRUD Operations

Add insert, delete, update, and select methods that build SQL statements and delegate to execute_update or execute_query.

class MySQLDatabase:
    # ... previous methods ...
    def insert(self, table, data):
        """Insert a row"""
        columns = ', '.join(data.keys())
        values = ', '.join(['%s'] * len(data))
        query = f"INSERT INTO {table} ({columns}) VALUES ({values})"
        return self.execute_update(query, tuple(data.values()))

    def delete(self, table, condition):
        """Delete rows matching condition"""
        query = f"DELETE FROM {table} WHERE {condition}"
        return self.execute_update(query)

    def update(self, table, data, condition):
        """Update rows matching condition"""
        set_clause = ', '.join([f"{k}=%s" for k in data.keys()])
        query = f"UPDATE {table} SET {set_clause} WHERE {condition}"
        return self.execute_update(query, tuple(data.values()))

    def select(self, table, condition=None):
        """Select rows, optionally with a WHERE clause"""
        if condition:
            query = f"SELECT * FROM {table} WHERE {condition}"
        else:
            query = f"SELECT * FROM {table}"
        return self.execute_query(query)

3. Parameterized Test Cases with Pytest

Define test data as a list of tuples, each containing the table name, insert payload, delete condition, and update payload.

test_data = [
    (
        "users",
        {"name": "John Doe", "email": "[email protected]"},
        "name='John Doe'",
        {"email": "[email protected]"}
    ),
    (
        "users",
        {"name": "Jane Doe", "email": "[email protected]"},
        "name='Jane Doe'",
        {"email": "[email protected]"}
    ),
]

Write a parametrized test that creates a MySQLDatabase instance, performs insert, query, update, and delete operations, and asserts the expected row counts.

import pytest
from database import MySQLDatabase

@pytest.mark.parametrize("table, insert_data, condition, update_data", test_data)
def test_database_operations(table, insert_data, condition, update_data):
    db = MySQLDatabase(host='localhost', user='root', password='password', db='testdb')
    db.connect()

    inserted_rows = db.insert(table, insert_data)
    assert inserted_rows == 1, "Insert failed"

    query = f"SELECT * FROM {table} WHERE {condition}"
    result = db.execute_query(query)
    assert json.loads(result), "Query failed"

    updated_rows = db.update(table, update_data, condition)
    assert updated_rows == 1, "Update failed"

    deleted_rows = db.delete(table, condition)
    assert deleted_rows == 1, "Delete failed"

    db.close()

4. Returning Data as JSON

The execute_query method already returns a JSON‑encoded string, allowing test code to consume the result directly without extra conversion.

5. Logging Integration with Loguru

Install Loguru and configure a rotating log file.

pip install loguru

from loguru import logger
logger.add("logs/{time}.log", rotation="1 day", compression="zip")

Use logger.info() at key points in the test workflow (start, API request, DB query, completion) to provide a clear execution trace.

6. Summary

By encapsulating MySQL operations, leveraging Pytest parameterization, returning query results as JSON, and adding structured logging, you can dramatically improve the readability, reusability, and maintainability of API automation tests that depend on database verification.