Python如何传入不同参数复用配置 批量调用Twitter API

出现仅返回最后一个用户数据的问题，和Twitter API权限、限流逻辑无关，是代码里的两个基础逻辑错误：

• 结果变量被循环覆盖：如果在循环外定义单个变量存储请求结果，每次拿到当前用户的关注列表后直接对该变量重新赋值，没有追加到统一的结果集合中，循环结束后变量自然只会保留最后一次赋值的内容，也就是最后一个用户的返回数据。
• 原始接口模板被污染：如果把end_point作为全局/类属性存储，循环时直接在原变量上执行replace赋值（即写end_point = end_point.replace("id", uid)），第一次替换后模板里的id占位符就会被替换成第一个用户ID，后续循环执行replace时找不到匹配的id子串，会导致URL生成异常。如果循环内的异常捕获逻辑有问题，出错后直接跳过前面的用户，最终也会只落到最后一个用户的请求上。

另外当前逻辑还有个隐藏缺陷：没有处理接口分页。Twitter v2的关注列表接口单次请求最多返回1000条数据，返回体带next_token时必须携带分页参数拉取剩余内容，就算单用户的关注列表也会拉取不完整。

核心原则是配置常量全程不修改，方法职责拆分，单用户请求和批量遍历逻辑解耦：

1. 固定配置为全局常量，绝对不在循环中修改，用str.format做占位符替换，避免replace误匹配其他位置的"id"字符

import time
import requests
from pymongo import MongoClient

# 接口模板固定为常量，全程不修改
FOLLOWING_ENDPOINT_TPL = "https://api.twitter.com/2/users/{user_id}/following"
# 固定请求头
HEADERS = {"Authorization": f"Bearer {barer_token}"}

2. 拆分独立的待查询用户ID拉取方法，不要把查库逻辑塞到请求循环里，避免重复查询或者Mongo游标异常

def get_pending_user_ids(users_col):
    # 查询所有未爬取关注列表的用户ID
    pending_users = users_col.find(
        {"following_crawled": {"$ne": True}},
        {"_id": 1}
    )
    return [str(u["_id"]) for u in pending_users]

3. 重构单用户请求方法，用户ID作为入参传入，内置分页拉取、429限流重试逻辑，不在单个请求方法里遍历所有用户

def fetch_single_user_following(user_id: str):
    user_following = []
    pagination_token = None
    while True:
        # 每次请求基于原始模板临时生成URL，不修改常量
        url = FOLLOWING_ENDPOINT_TPL.format(user_id=user_id)
        params = {"max_results": 1000}
        if pagination_token:
            params["pagination_token"] = pagination_token
        
        while True:
            resp = requests.get(url, headers=HEADERS, params=params)
            if resp.status_code == 200:
                data = resp.json()
                # 处理接口返回的业务错误
                if "errors" in data:
                    raise Exception(f"接口返回错误: {data['errors']}")
                user_following.extend(data.get("data", []))
                # 提取分页token判断是否需要继续拉取
                pagination_token = data.get("meta", {}).get("next_token")
                break
            elif resp.status_code == 429:
                # 先等待1秒做快速重试
                time.sleep(1)
                retry_resp = requests.get(url, headers=HEADERS, params=params)
                if retry_resp.status_code != 429:
                    resp = retry_resp
                    continue
                # 快速重试失败则读取限流重置时间计算等待时长
                reset_ts = int(resp.headers.get("x-rate-limit-reset", time.time() + 900))
                sleep_sec = max(reset_ts - time.time(), 0)
                print(f"触发接口限流，等待{sleep_sec}秒后重试")
                time.sleep(sleep_sec)
            else:
                raise Exception(f"请求异常，状态码: {resp.status_code}, 响应内容: {resp.text}")
        
        # 没有下一页时终止当前用户的拉取循环
        if not pagination_token:
            break
    return user_following

4. 主流程实现批量遍历，每处理完一个用户立刻落库并标记爬取状态，避免程序中途崩溃导致所有数据丢失

def main():
    # 初始化MongoDB连接
    tweets_client = MongoClient(mongodb_tweets_conn)
    tweets_col = tweets_client[mongodb_tweets_db][mongodb_tweets_col]
    users_client = MongoClient(mongodb_users_conn)
    users_col = users_client[mongodb_users_db][mongodb_users_col]

    # 一次性拉取所有待处理用户ID
    pending_uids = get_pending_user_ids(users_col)
    for uid in pending_uids:
        print(f"正在拉取用户{uid}的关注列表")
        try:
            following_list = fetch_single_user_following(uid)
            # 批量写入关注关系数据，根据实际业务需求调整存储结构
            if following_list:
                tweets_col.insert_many([
                    {"from_user_id": uid, "to_user_id": f["id"]}
                    for f in following_list
                ])
            # 标记当前用户爬取完成
            users_col.update_one({"_id": uid}, {"$set": {"following_crawled": True}})
        except Exception as e:
            print(f"拉取用户{uid}失败: {str(e)}")
            # 标记爬取失败，后续可做重试
            users_col.update_one({"_id": uid}, {"$set": {"following_crawled": False, "err_msg": str(e)}})
            continue

if __name__ == "__main__":
    main()

• 禁止在循环中修改原始配置常量，所有动态请求参数都基于常量临时生成
• 批量结果必须用列表、字典这类支持追加的结构存储，不要用单个变量反复赋值覆盖
• 429重试时一定要保留当前请求的全量参数（包括分页token），不要重试时跳页或者跳过当前用户
• 不要等所有用户爬完再统一写库，单用户处理完立刻落库，降低异常带来的数据丢失风险

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