Python拉取QuiverQuant数据时如何无需重复代码为类变量赋值

你可以通过建立URL和对应类存储变量的映射关系实现自动匹配存储，无需编写大量重复判断逻辑，修改后的代码如下：

import json
import requests

# 接口地址定义保留原有逻辑
house = "https://api.quiverquant.com/beta/live/housetrading"
congress = "https://api.quiverquant.com/beta/live/congresstrading"
senate = "https://api.quiverquant.com/beta/live/senatetrading"
contracts = "https://api.quiverquant.com/beta/live/govcontractsall"
quarterly_contracts = "https://api.quiverquant.com/beta/live/govcontracts"
corp_lobbying = "https://api.quiverquant.com/beta/live/lobbying"
off_exchange = "https://api.quiverquant.com/beta/live/offexchange"
congress_ticker = "https://api.quiverquant.com/beta/historical/congresstrading/{ticker}"

class GetData():
    # 你已定义的存储类变量全部保留
    data = []
    house_data = []
    congress_data = []
    senate_data = []
    contracts_data = []
    quarterly_contracts_data = []
    corp_lobbying_data = []
    off_exchange_data = []
    congress_ticker_data = []
    
    # 新增URL与对应存储字段的映射表
    url_to_field = {
        house: "house_data",
        congress: "congress_data",
        senate: "senate_data",
        contracts: "contracts_data",
        quarterly_contracts: "quarterly_contracts_data",
        corp_lobbying: "corp_lobbying_data",
        off_exchange: "off_exchange_data",
        congress_ticker: "congress_ticker_data"
    }

    def __init__(self, sector):
        self.sector = sector
        # 初始化时匹配当前接口对应的存储字段名
        self.target_field = self.url_to_field.get(sector)
        if not self.target_field:
            raise ValueError(f"未找到接口 {sector} 对应的存储配置")

    def get_data(self): 
        url = self.sector
        headers = {
            'accept': 'application/json',
            'X-CSRFToken': 'TyTJwjuEC7VV7mOqZ622haRaaUr0x0Ng4nrwSRFKQs7vdoBcJlK9qjAS69ghzhFu',
            'Authorization': 'Token 替换为你自己的授权Token'
        }
        r = requests.get(url, headers=headers) 
        parsed_data = json.loads(r.content)
        # 将数据写入对应的类变量中
        getattr(GetData, self.target_field).append(parsed_data)
        print(f'已从 {url} 拉取数据，存入 {self.target_field}')
    
    def print_data(self):
        try:
            self.get_data()
            # 打印当前接口对应的存储数据
            print(getattr(GetData, self.target_field))
        except Exception as e:
            print(f'操作失败: {str(e)}')

# 原有调用方式不变
house_info = GetData(house)
congress_info = GetData(congress)
senate_info = GetData(senate)
congress_info.get_data()
# house_info.get_data()
# senate_info.print_data()

核心改动说明 #

• 新增url_to_field映射表，所有接口和对应存储字段的关系统一维护，后续新增接口只需在表中添加一条记录即可，无需修改业务逻辑
• 初始化时自动匹配当前接口对应的存储字段，拉取数据后通过getattr动态写入对应类变量，各板块数据完全独立，不会出现混淆
• 修复了原print_data方法中调用不存在的piped_data属性的问题

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