Python用Flask封装API及调用详解

from flask import Flask, request, Response, jsonify, abort
import json

app = Flask(__name__)

def main(query, k):
    """业务逻辑处理函数
    
    Args:
        query (str): 查询文本
        k (int): 返回结果数量
    
    Returns:
        dict: 包含查询参数和结果的字典
    """
    return {"query": query, "k": k, "status": "success"}

@app.route('/api/search', methods=['GET'])
def handle_get():
    """处理GET请求
    
    参数通过URL查询字符串传递:
    - query: 必需，字符串类型
    - k: 可选，整数类型，默认为1
    
    Returns:
        Response: JSON格式的响应
    """
    # 获取并验证参数
    query = request.args.get("query")
    if not query:
        abort(400, description="query参数不能为空")
    
    try:
        k = int(request.args.get("k", 1))  # 默认为1
    except ValueError:
        abort(400, description="k必须是整数")
    
    # 调用业务逻辑
    result = main(query, k)
    
    # 返回JSON响应
    return jsonify(result)

import requests

# 构造URL参数
params = {
    'query': 'Python Flask',
    'k': 3
}

# 发送GET请求
response = requests.get(
    url="http://127.0.0.1:8000/api/search",
    params=params
)

# 处理响应
if response.status_code == 200:
    print(response.json())
else:
    print(f"请求失败，状态码: {response.status_code}, 错误: {response.text}")

@app.route('/api/form-submit', methods=['POST'])
def handle_post_form():
    """处理表单POST请求
    
    参数通过表单数据传递:
    - query: 必需，字符串类型
    - k: 可选，整数类型，默认为1
    
    Returns:
        Response: JSON格式的响应
    """
    # 获取并验证参数
    query = request.form.get("query")
    if not query:
        abort(400, description="query参数不能为空")
    
    try:
        k = int(request.form.get("k", 1))  # 默认为1
    except ValueError:
        abort(400, description="k必须是整数")
    
    # 调用业务逻辑
    result = main(query, k)
    
    # 返回JSON响应
    return jsonify(result)

import requests

# 准备表单数据
form_data = {
    'query': '表单提交示例',
    'k': 2
}

# 发送POST请求
response = requests.post(
    url="http://127.0.0.1:8000/api/form-submit",
    data=form_data
)

# 处理响应
if response.status_code == 200:
    print(response.json())
else:
    print(f"请求失败，状态码: {response.status_code}, 错误: {response.text}")

@app.route('/api/json-submit', methods=['POST'])
def handle_post_json():
    """处理JSON POST请求
    
    参数通过JSON格式传递:
    - query: 必需，字符串类型
    - k: 可选，整数类型，默认为1
    
    Returns:
        Response: JSON格式的响应
    """
    # 验证请求内容类型
    if not request.is_json:
        abort(415, description="请求必须是JSON格式")
    
    # 获取JSON数据
    data = request.get_json()
    
    # 验证参数
    query = data.get("query")
    if not query:
        abort(400, description="query参数不能为空")
    
    try:
        k = int(data.get("k", 1))  # 默认为1
    except ValueError:
        abort(400, description="k必须是整数")
    
    # 调用业务逻辑
    result = main(query, k)
    
    # 返回JSON响应
    return jsonify(result)

import requests

# 准备JSON数据
json_data = {
    'query': 'JSON数据示例',
    'k': 4
}

# 发送POST请求
response = requests.post(
    url="http://127.0.0.1:8000/api/json-submit",
    json=json_data,  # 自动设置Content-Type为application/json
    headers={'Accept': 'application/json'}  # 明确要求JSON响应
)

# 处理响应
if response.status_code == 200:
    print(response.json())
else:
    print(f"请求失败，状态码: {response.status_code}, 错误: {response.text}")

@app.errorhandler(400)
def bad_request(error):
    return jsonify({
        'status': 'error',
        'message': error.description
    }), 400

@app.errorhandler(404)
def not_found(error):
    return jsonify({
        'status': 'error',
        'message': '资源不存在'
    }), 404

@app.errorhandler(415)
def unsupported_media_type(error):
    return jsonify({
        'status': 'error',
        'message': error.description
    }), 415

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000, debug=True)