Python SQL连接：零基础转存pytest响应数据至SQL表全方法指导

Hey there! Let's walk through everything you need to know as a beginner to handle pytest response data—storing it, loading it, and saving it to an SQL database. I'll break it down into simple, actionable steps, starting with the basics of connecting Python to SQL.

Before diving into pytest data handling, you need to know how Python talks to SQL databases. Here are the most common tools for beginners:

a. SQLite3 (Built-in, No Installation Needed) #

Perfect for learning and small projects—no separate database server required, it uses local files.

import sqlite3

# Connect to a database (creates the file if it doesn't exist)
with sqlite3.connect('my_test_db.db') as conn:
    cursor = conn.cursor()
    # Run your SQL commands here
    cursor.execute("SELECT * FROM some_table")
    results = cursor.fetchall()
# The connection closes automatically when the `with` block ends

b. MySQL/MariaDB: mysql-connector-python #

For MySQL databases. First install it:

pip install mysql-connector-python

Connection example:

import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    user="your_username",
    password="your_password",
    database="your_db_name"
)
cursor = conn.cursor()

# Don't forget to commit changes and close when done
conn.commit()
conn.close()

c. PostgreSQL: psycopg2-binary #

For PostgreSQL. Install first:

pip install psycopg2-binary

Connection example:

import psycopg2

conn = psycopg2.connect(
    dbname="your_db_name",
    user="your_username",
    password="your_password",
    host="localhost",
    port="5432"
)
cursor = conn.cursor()
conn.commit()
conn.close()

d. SQLAlchemy (ORM, For Easier Management) #

ORM stands for Object-Relational Mapping—it lets you work with database tables like Python classes, no raw SQL required (though you can still use it if you want). Great for larger projects. Install it:

pip install sqlalchemy

Basic setup example:

from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Create an engine (connect to your database)
engine = create_engine('sqlite:///my_orm_db.db')  # For other DBs: `postgresql://user:pass@localhost/dbname`
Base = declarative_base()

# Define a "model" (maps to a database table)
class ApiResponse(Base):
    __tablename__ = 'api_responses'
    id = Column(Integer, primary_key=True, autoincrement=True)
    endpoint = Column(String, nullable=False)
    status_code = Column(Integer, nullable=False)
    response_body = Column(String, nullable=False)

# Create all tables (runs once)
Base.metadata.create_all(engine)

# Create a session to interact with the DB
Session = sessionmaker(bind=engine)
session = Session()

Before storing anything, you need to get and structure your pytest response data. Let's use the requests library to send API calls (install it first: pip install requests):

import pytest
import requests
import json

def test_api_call():
    # Send a GET request
    response = requests.get("https://jsonplaceholder.typicode.com/todos/1")
    
    # Extract the data we want to store
    structured_data = {
        "endpoint": response.url,
        "status_code": response.status_code,
        "response_body": json.dumps(response.json())  # Convert JSON to string for DB storage
    }
    
    return structured_data

Note: We use json.dumps() to turn the JSON response into a string because most SQL databases can't store raw JSON objects directly. Later, we'll use json.loads() to convert it back to a Python dict.

This is straightforward if you want to write explicit SQL commands.

Step 1: Initialize Your Database Table #

Use a pytest fixture to set up the table once before all tests run:

import sqlite3
import pytest

@pytest.fixture(scope="session", autouse=True)
def setup_database():
    # Connect and create table
    with sqlite3.connect('test_responses.db') as conn:
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS api_responses (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                endpoint TEXT NOT NULL,
                status_code INTEGER NOT NULL,
                response_body TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()

Step 2: Store Response Data in the Test #

def test_store_response_raw_sql():
    response_data = test_api_call()  # Reuse the function from earlier
    
    with sqlite3.connect('test_responses.db') as conn:
        cursor = conn.cursor()
        # Use parameterized queries to avoid SQL injection!
        cursor.execute('''
            INSERT INTO api_responses (endpoint, status_code, response_body)
            VALUES (?, ?, ?)
        ''', (response_data["endpoint"], response_data["status_code"], response_data["response_body"]))
        conn.commit()

Step 3: Load Data from the Database #

def test_load_response_raw_sql():
    with sqlite3.connect('test_responses.db') as conn:
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM api_responses')
        rows = cursor.fetchall()
        
        # Convert rows to a list of dicts for easy use
        loaded_responses = []
        for row in rows:
            loaded_responses.append({
                "id": row[0],
                "endpoint": row[1],
                "status_code": row[2],
                "response_body": json.loads(row[3]),  # Convert string back to JSON
                "created_at": row[4]
            })
        
        assert len(loaded_responses) > 0
        print(loaded_responses[0])

This is cleaner for larger projects—no raw SQL needed once the model is defined.

Step 1: Use SQLAlchemy Fixture #

import pytest
from sqlalchemy.orm import sessionmaker
from your_module import engine, Base, ApiResponse  # Import your ORM setup

@pytest.fixture(scope="session")
def db_session():
    # Create tables
    Base.metadata.create_all(engine)
    session = Session(bind=engine)
    yield session
    # Cleanup after tests
    session.close()

Step 2: Store Response Data #

def test_store_response_orm(db_session):
    response_data = test_api_call()
    
    # Create an instance of our model
    new_response = ApiResponse(
        endpoint=response_data["endpoint"],
        status_code=response_data["status_code"],
        response_body=response_data["response_body"]
    )
    
    # Add and commit to the database
    db_session.add(new_response)
    db_session.commit()

Step 3: Load Data #

def test_load_response_orm(db_session):
    # Query all responses
    responses = db_session.query(ApiResponse).all()
    
    assert len(responses) > 0
    # Parse the response body back to JSON
    first_response = responses[0]
    parsed_body = json.loads(first_response.response_body)
    print(f"Loaded response from {first_response.endpoint}: {parsed_body}")

If you just need to share response data between pytest tests (not permanent storage), use a session-scoped fixture:

import pytest
import requests

@pytest.fixture(scope="session")
def shared_api_response():
    # Send the request once, reuse the response across all tests
    return requests.get("https://jsonplaceholder.typicode.com/todos/1")

# Use the shared response in tests
def test_check_status_code(shared_api_response):
    assert shared_api_response.status_code == 200

def test_check_user_id(shared_api_response):
    assert shared_api_response.json()["userId"] == 1

This is great for avoiding redundant API calls and speeding up tests—no database required!

• Avoid SQL Injection: Always use parameterized queries (like the ? placeholders in raw SQL) instead of string concatenation. Never do cursor.execute(f"INSERT INTO ... VALUES ({data})")—it's a huge security risk.
• Manage Connections: Use the with context manager to auto-close database connections, or remember to call conn.close() manually to avoid connection leaks.
• Serialization: Always convert JSON responses to strings before storing in SQL, and parse them back when loading.
• Fixture Scope: Use scope="session" for database setup fixtures so you don't recreate tables every time a test runs—it saves time.

内容的提问来源于stack exchange，提问作者Pianca