Flask处理curl POST请求时如何校验重复IP与主机名过滤重复数据

原校验逻辑失效原因 #

你之前逐行读JSON文本做字符串匹配的方案不可行：格式化后的JSON文件每行包含缩进、引号、逗号等格式字符，和你拼接的IP 主机名字符串结构完全不同，永远无法匹配到重复项。同时原逻辑没有覆盖批量提交内部的重复场景，也没有做写入前的全量校验，会直接把脏数据写入文件。

可直接落地的实现代码 #

实现逻辑遵循「先全量校验，后写入文件」的原则，一旦发现重复、格式错误直接返回400状态码，不会产生脏数据：

from flask import request, jsonify
import json
import ipaddress

# 补全你的路由装饰器，和现有服务的路由规则保持一致即可
@app.route('/hosts', methods=['POST'])
def put_hosts():
    content = request.get_json(silent=True)
    # 基础格式校验
    if not isinstance(content, list):
        return jsonify({"error": "请求体必须为JSON数组格式"}), 400

    # 加载存量配置，兼容文件不存在、格式损坏的场景
    try:
        with open('example.json', 'r', encoding='utf-8') as config_file:
            stored_data = json.load(config_file)
    except (json.JSONDecodeError, FileNotFoundError):
        stored_data = []

    # 预收集存量数据的唯一标识，提升校验效率
    exist_hostnames = {item['hostname'] for item in stored_data if 'hostname' in item}
    exist_ips = {item['address'] for item in stored_data if 'address' in item}
    exist_routes = [ipaddress.ip_network(item['routes'], strict=False) for item in stored_data if 'routes' in item]

    # 临时集合存储本次提交的条目，校验批量内部重复
    batch_hostnames = set()
    batch_ips = set()

    for entry in content:
        # 校验必填字段完整性
        if not all(key in entry for key in ('address', 'hostname', 'routes')):
            return jsonify({"error": f"条目缺失必填字段: {entry}"}), 400
        
        hostname = entry['hostname']
        ip_addr = entry['address']
        route = entry['routes']

        # 主机名校验：批量内部 + 存量数据
        if hostname in batch_hostnames or hostname in exist_hostnames:
            return jsonify({"error": f"主机名重复: {hostname}"}), 400
        
        # IP地址校验：批量内部 + 存量数据
        if ip_addr in batch_ips or ip_addr in exist_ips:
            return jsonify({"error": f"IP地址重复: {ip_addr}"}), 400
        
        # 路由网段格式与重叠校验
        try:
            new_route = ipaddress.ip_network(route, strict=False)
        except ValueError:
            return jsonify({"error": f"无效路由网段格式: {route}"}), 400
        for r in exist_routes:
            if new_route.overlaps(r):
                return jsonify({"error": f"路由网段{route}与已有网段{r}重叠"}), 400

        # 单条校验通过，加入临时集合用于后续条目校验
        batch_hostnames.add(hostname)
        batch_ips.add(ip_addr)
        exist_routes.append(new_route)

    # 全量校验通过后再写入文件，保证原子性
    stored_data.extend(content)
    with open('example.json', 'w', encoding='utf-8') as config_file:
        json.dump(stored_data, config_file, sort_keys=True, indent=4, ensure_ascii=False)
    
    return jsonify(stored_data), 201

校验规则说明 #

• 主机名唯一性：覆盖存量数据重复、本次批量提交内部重复两种场景
• IP地址唯一性：和主机名校验逻辑一致，出现重复直接拦截，不会写入任何数据
• 路由重叠校验：使用Python标准库ipaddress实现CIDR网段重叠判断，无需安装第三方依赖
• 异常兼容：自动处理存储文件不存在、JSON格式损坏、请求字段缺失、路由格式非法等边界场景
• 原子写入：所有条目全部校验通过后才执行文件写入，避免出现部分数据写入、文件损坏的问题

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