基于CSV基线数据验证REST服务JSON响应的技术咨询

1. Python Script (Flexible & Customizable) #

This is my go-to approach because it lets you tailor the comparison exactly to your needs. Here's a step-by-step breakdown:

Steps: #

• Fetch the REST API response: Use the requests library to call your endpoint and parse the JSON data.
• Load the CSV baseline: Use pandas to read the exported Oracle CSV into a DataFrame (it handles data type conversion and normalization well).
• Normalize both datasets: Ensure column names match (e.g., if your JSON uses first_name and the CSV uses First Name, rename columns to align). Also, cast data types consistently (like converting id from string to integer if needed).
• Compare records: Check for missing/extra records, and verify that all field values match between the two datasets.

Sample Code: #

import requests
import pandas as pd

# Step 1: Fetch API data
api_url = "https://your-rest-service-endpoint.com/users"
response = requests.get(api_url)
api_data = response.json()
api_df = pd.DataFrame(api_data)

# Step 2: Load CSV baseline (adjust path as needed)
csv_df = pd.read_csv("oracle_baseline.csv")

# Step 3: Normalize columns (example if CSV uses different names)
csv_df = csv_df.rename(columns={
    "First Name": "first_name",
    "Last Name": "last_name",
    "Avatar": "avatar"
})

# Ensure ID is integer in both
api_df["id"] = api_df["id"].astype(int)
csv_df["id"] = csv_df["id"].astype(int)

# Step 4: Compare datasets
# Check for missing records in API
missing_in_api = csv_df[~csv_df["id"].isin(api_df["id"])]
if not missing_in_api.empty:
    print("Missing records in API response:", missing_in_api)

# Check for extra records in API
extra_in_api = api_df[~api_df["id"].isin(csv_df["id"])]
if not extra_in_api.empty:
    print("Extra records in API response:", extra_in_api)

# Check for field mismatches
merged_df = pd.merge(api_df, csv_df, on="id", suffixes=("_api", "_csv"), how="inner")
mismatches = merged_df[
    (merged_df["first_name_api"] != merged_df["first_name_csv"]) |
    (merged_df["last_name_api"] != merged_df["last_name_csv"]) |
    (merged_df["avatar_api"] != merged_df["avatar_csv"])
]

if not mismatches.empty:
    print("Field mismatches found:", mismatches)
else:
    print("All records match successfully!")

2. API Testing Tools (Postman/Newman) #

If you prefer a no-code/low-code approach, tools like Postman work great for validating API responses against CSV data:

Steps: #

1. Import your CSV baseline into Postman as a data source (go to "Runner" > "Select File").
2. Create a request to your REST endpoint.
3. Write test scripts in the "Tests" tab to compare each field from the API response to the corresponding row in the CSV.

Sample Postman Test Script: #

// Get current row from CSV data
const csvRow = pm.iterationData.toObject();

// Parse API response
const apiResponse = pm.response.json();

// Validate each field
pm.test(`ID matches for row ${pm.iteration}`, () => {
    pm.expect(apiResponse.id).to.eql(parseInt(csvRow.id));
});

pm.test(`First name matches for row ${pm.iteration}`, () => {
    pm.expect(apiResponse.first_name).to.eql(csvRow.first_name);
});

pm.test(`Last name matches for row ${pm.iteration}`, () => {
    pm.expect(apiResponse.last_name).to.eql(csvRow.last_name);
});

pm.test(`Avatar matches for row ${pm.iteration}`, () => {
    pm.expect(apiResponse.avatar).to.eql(csvRow.avatar);
});

Run this via Postman Runner, and it will iterate through each row in the CSV to validate the API response.

3. Direct Database Comparison (Skip the CSV Middleman) #

Instead of relying on an exported CSV (which can have formatting errors or be outdated), you can connect directly to your Oracle database from a script to fetch the baseline data. This is more reliable for ongoing validation.

Sample Python Code with Oracle Connection: #

import requests
import pandas as pd
import cx_Oracle

# Fetch API data
api_url = "https://your-rest-service-endpoint.com/users"
response = requests.get(api_url)
api_df = pd.DataFrame(response.json())

# Connect to Oracle DB (adjust credentials and query as needed)
connection = cx_Oracle.connect(
    user="your_username",
    password="your_password",
    dsn="your_oracle_dsn"
)

# Query baseline data from Oracle
query = """
SELECT id, first_name, last_name, avatar 
FROM your_table 
ORDER BY id
"""
db_df = pd.read_sql(query, connection)
connection.close()

# Compare datasets (same as step 4 in the first approach)
merged_df = pd.merge(api_df, db_df, on="id", suffixes=("_api", "_db"), how="inner")
mismatches = merged_df[
    (merged_df["first_name_api"] != merged_df["first_name_db"]) |
    (merged_df["last_name_api"] != merged_df["last_name_db"]) |
    (merged_df["avatar_api"] != merged_df["avatar_db"])
]

if not mismatches.empty:
    print("Mismatches found:", mismatches)
else:
    print("API response matches database baseline perfectly!")

Key Considerations #

• Data Type Consistency: Ensure fields like id are the same type (integer vs string) in both datasets before comparing.
• Record Order: If your API doesn't return records in a fixed order, avoid comparing based on row index—use unique identifiers like id instead.
• Edge Cases: Handle null values, empty strings, and special characters (some CSV exports might escape these differently).
• Automation: For regular validation, wrap the script in a cron job or CI/CD pipeline (like Jenkins) to run checks automatically.

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