Python接口开发最实用的20个Skills:从传统手工到AI智能化的进阶之路

import requestssession = requests.Session()# 登录获取tokenlogin_resp = session.post("http://api.example.com/login", json={"user": "test", "pwd": "123456"})token = login_resp.json()["token"]# 携带token调用业务接口headers = {"Authorization": f"Bearer {token}"}resp = session.get("http://api.example.com/user/profile", headers=headers, timeout=5)

import pytest@pytest.fixturedef db_connection():    conn = create_db_connection()    yield conn    conn.close()  # 测试结束后自动关闭连接@pytest.mark.parametrize("user_id, expected_status", [(1, 200), (999, 404)])def test_get_user(db_connection, user_id, expected_status):    resp = requests.get(f"http://api.example.com/users/{user_id}")    assert resp.status_code == expected_status

import jsonschemaschema = {    "type": "object",    "properties": {"code": {"type": "integer"}, "data": {"type": "object"}},    "required": ["code", "data"]}resp_json = {"code": 200, "data": {"name": "test"}}jsonschema.validate(resp_json, schema)  # 校验通过则无异常，否则抛出ValidationError

import pymysqlconn = pymysql.connect(host="localhost", user="root", password="password", database="test_db")cursor = conn.cursor()cursor.execute("SELECT id FROM users WHERE status=1 LIMIT 1")user_id = cursor.fetchone()[0]  # 获取有效用户ID用于接口测试cursor.close()conn.close()‍

class BaseApi:    def __init__(self, base_url):        self.base_url = base_url        self.session = requests.Session()    def send_request(self, method, url, **kwargs):        full_url = self.base_url + url        try:            resp = self.session.request(method, full_url, **kwargs)            logging.info(f"请求: {method}{full_url}, 响应: {resp.text}")            return resp        except Exception as e:            logging.error(f"请求失败: {e}")            raise

from faker import Fakerfake = Faker("zh_CN")name = fake.name()  # 生成随机中文姓名email = fake.email()  # 生成随机邮箱

安装allure-pytest：pip install allure-pytest执行测试生成数据：pytest --alluredir=./report_data生成报告：allure generate ./report_data -o ./report --clean

pipeline {    agent any    stages {        stage("拉取代码") {            steps {                git url: "http://git.example.com/test_project.git"            }        }        stage("执行测试") {            steps {                sh "pip install -r requirements.txt"                sh "pytest test_api.py --alluredir=./report_data"            }        }        stage("生成报告") {            steps {                sh "allure generate ./report_data -o ./report --clean"            }        }    }}‍

from openai import OpenAIclient = OpenAI(api_key="your_api_key")swagger_json = {    "path": "/api/users",    "method": "POST",    "parameters": {"name": {"type": "string", "required": True}, "age": {"type": "integer"}}}prompt = f"""根据以下接口信息生成测试用例：接口路径: {swagger_json['path']}方法: {swagger_json['method']}参数: {swagger_json['parameters']}要求：包含1个正向用例（参数正确）和3个异常用例（参数缺失、类型错误、超长）。输出JSON格式：[{"description": "xxx", "request": {}, "expected_status": 200}]"""response = client.chat.completions.create(model="gpt-4", messages=[{"role": "user", "content": prompt}])test_cases = json.loads(response.choices[0].message.content)

response_schema = {"code": 200, "data": {"user_id": 123, "name": "test"}}prompt = f"""根据以下响应数据生成Python断言：{response_schema}要求：检查code是否为200，data中是否包含user_id和name，user_id是否为整数。"""response = client.chat.completions.create(model="gpt-4", messages=[{"role": "user", "content": prompt}])assert_code = response.choices[0].message.content# 输出示例：assert resp_json["code"] == 200; assert "user_id" in resp_json["data"]; assert isinstance(resp_json["data"]["user_id"], int)

import timedef ai_retry(func, max_retries=3):    for i in range(max_retries):        try:            return func()        except Exception as e:            if "网络超时" in str(e) and i < max_retries - 1:                delay = 2 ** i  # 指数退避                time.sleep(delay)            else:                raise# 使用示例：ai_retry(lambda: requests.get("http://api.example.com/status"))

prompt = "生成10个中国用户信息，包含姓名、身份证号、手机号，输出JSON格式"response = client.chat.completions.create(model="gpt-4", messages=[{"role": "user", "content": prompt}])test_data = json.loads(response.choices[0].message.content)

import requestsswagger_resp = requests.get("http://api.example.com/v2/api-docs")swagger_json = swagger_resp.json()paths = swagger_json["paths"]  # 获取所有接口路径for path, methods in paths.items():    for method, detailsinmethods.items():        print(f"接口: {method.upper()} {path}, 参数: {details.get('parameters', [])}")

@pytest.fixturedef test_user():    # 创建测试用户    user_id = create_user_in_db()    yield user_id    # 删除测试用户    delete_user_from_db(user_id)def test_update_user(test_user):    user_id = test_user    resp = requests.put(f"http://api.example.com/users/{user_id}", json={"name": "new_name"})    assert resp.status_code == 200

import loggingimport uuidlogging.basicConfig(level=logging.INFO, format="%(asctime)s - %(trace_id)s - %(message)s")def send_request_with_trace(url, data):    trace_id = str(uuid.uuid4())    logging.info(f"请求: {url}, 数据: {data}, trace_id: {trace_id}")    resp = requests.post(url, json=data, headers={"X-Trace-Id": trace_id})    logging.info(f"响应: {resp.text}, trace_id: {trace_id}")    return resp

from locust import HttpUser, taskclass ApiUser(HttpUser):    @task    def get_user_profile(self):        self.client.get("/api/user/profile")    @task(3)    def search_products(self):        self.client.post("/api/products/search", json={"keyword": "手机"})

# SQL注入测试sql_injection_payload = "1' OR '1'='1"resp = requests.get(f"http://api.example.com/users/{sql_injection_payload}")assert resp.status_code == 400  # 期望返回400错误，说明接口有SQL注入防护

import osenv = os.environ.get("ENV", "test")  # 默认测试环境config = {    "test": {"base_url": "http://test.api.example.com"},    "prod": {"base_url": "http://api.example.com"}}base_url = config[env]["base_url"]

import pandas as pd# 读取Allure生成的测试结果CSVdf = pd.read_csv("test_results.csv")# 统计各接口的失败次数failure_count = df[df["status"] == "failed"]["api_path"].value_counts()print(failure_count)# 输出示例：# /api/orders    15# /api/users     8# /api/products  3‍