Tweepy StreamingClient自定义__init__传bearer_token报错求助

你的TweetStreamer子类__init__方法没有定义bearer_token参数，也未接收额外关键字参数，但创建实例时传入了该参数，导致Python报错；同时父类StreamingClient的__init__需要bearer_token作为必传参数，你当前调用super().__init__()时未传递该参数，即使没触发此错误，也会导致父类初始化失败。

有两种修改方式，可根据需求选择：

方式1：显式声明bearer_token参数 #

直接在子类__init__中添加bearer_token参数，并传递给父类初始化方法：

import datetime
import tweepy

def check_path_exists(path):
    # 假设该函数已实现路径校验逻辑
    return path

class TweetStreamer(tweepy.StreamingClient):

    def __init__(self, 
                 bearer_token: str,
                 out_path: str,
                 kill_time: int = 59,
                 time_unit: str = 'minutes'):
        '''
        添加自定义参数
        '''
        out_path = check_path_exists(out_path)
        self.outfile = out_path 

        if time_unit not in ['seconds', 'minutes']:
            raise ValueError(f'time_unit must be either `minutes` or `seconds`.')

        if time_unit == 'minutes':
            self.kill_time = datetime.timedelta(seconds=60*kill_time)
        else:
            self.kill_time = datetime.timedelta(seconds=kill_time)

        # 将bearer_token传递给父类初始化
        super().__init__(bearer_token=bearer_token)

    def on_data(self, data):
        '''
        1. 清理返回的推文对象
        2. 写入文件
        '''
        with open(self.outfile, 'ab') as o:
            o.write(data + b'\n')

方式2：用**kwargs兼容父类所有参数 #

通过**kwargs接收所有额外关键字参数，包括bearer_token及父类的其他可选参数（如wait_on_rate_limit），扩展性更强：

import datetime
import tweepy

def check_path_exists(path):
    # 假设该函数已实现路径校验逻辑
    return path

class TweetStreamer(tweepy.StreamingClient):

    def __init__(self, 
                 out_path: str,
                 kill_time: int = 59,
                 time_unit: str = 'minutes',
                 **kwargs):
        '''
        添加自定义参数，兼容父类所有关键字参数
        '''
        out_path = check_path_exists(out_path)
        self.outfile = out_path 

        if time_unit not in ['seconds', 'minutes']:
            raise ValueError(f'time_unit must be either `minutes` or `seconds`.')

        if time_unit == 'minutes':
            self.kill_time = datetime.timedelta(seconds=60*kill_time)
        else:
            self.kill_time = datetime.timedelta(seconds=kill_time)

        # 将所有额外关键字参数传递给父类
        super().__init__(**kwargs)

    def on_data(self, data):
        '''
        1. 清理返回的推文对象
        2. 写入文件
        '''
        with open(self.outfile, 'ab') as o:
            o.write(data + b'\n')

两种修改后的代码，都可以用你原来的方式创建实例：

streamer = TweetStreamer(bearer_token=bearer_token, out_path='test2.json')

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