Однажды я столкнулся с ограничениями скорости Twitter API, и тогда я захотел автоматизировать это.

Мне пришлось научить объект, как обернуть Tweepy API каким-то сонным Python.

Часть этого кода является частной собственностью, однако вы можете использовать часть этого для репликации этого решения.

Часть кода, упомянутого здесь, может быть немного продвинутой, но приятно видеть код, с которым вы, возможно, не совсем знакомы.

class TwitterAPIProxy(MagicObject2):
    def __init__(self, consumer_key=None, consumer_secret=None, access_token=None, access_token_secret=None, logger=None):
        self.consumer_key = consumer_key
        self.consumer_secret = consumer_secret
        self.access_token = access_token
        self.access_token_secret = access_token_secret
        self.logger = logger
        self.calls_count = 0
        self.start_time = time.time()
        self.rate_limit = os.environ.get('twitter_rate_limit')
        assert self.rate_limit and (isinstance(self.rate_limit, str)) and (len(self.rate_limit) > 0), 'Missing rate_limit.'
        self.rate_limit = self.rate_limit.split(':')
        assert (isinstance(self.rate_limit, list)) and (len(self.rate_limit) == 2), 'Missing proper rate_limit.'
        self.velocity = int(self.rate_limit[0]) / int(self.rate_limit[-1])

        try:
            self.auth = tweepy.OAuthHandler(self.consumer_key, self.consumer_secret)
            self.auth.set_access_token(self.access_token, self.access_token_secret)
            self.api = tweepy.API(self.auth)
        except:
            msg = 'Problem connecting to the twitter?'
            if (logger):
                logger.exception(msg)
            sys.exit()


    def __call__(self,*args,**kwargs):
        self.n = [n for n in self.n if (n != '__iter__')]
        if (len(self.n) > 0):
            self.calls_count += 1
            et = time.time() - self.start_time
            v = self.calls_count / et
            if (v > self.velocity):
                while (v > self.velocity):
                    if (self.logger):
                        self.logger.info('Sleeping due to twitter rate limits. (v is {} of {})'.format(v, self.velocity))
                    time.sleep(1)
                    et = time.time() - self.start_time
                    v = self.calls_count / et
            method = self.n.pop()
            s = 'self.api.%s(*args,**kwargs)' % (method)
            if (self.logger):
                self.logger.info('{} {}'.format(self.__class__, s))
            return eval(s)
        return None

Главное в этом коде - возможность засыпать всякий раз, когда скорость превышает 1.0, потому что Twitter разрешает 900 вызовов за 900 секунд, и это 1: 1, что также равно 1.0, не вдаваясь в математику.

Этот объект действует как оболочка для объекта API Tweepy.

У меня есть некоторые работы по отладке этого интерфейса, однако общие черты здесь, и они, похоже, работают в большинстве случаев.

Я обычно создаю ботов Twitter, которые работают в фоновом режиме 24 часа в сутки, 7 дней в неделю, и большую часть времени я не достигал пределов скорости, но недавно я начал искать методы, которые я мог бы использовать, чтобы добавить больше подписчиков в свою учетную запись Twitter, и это означало, что делать много вызовов API Twitter. Вы можете избежать ограничений скорости, если спите по 1 секунде для каждого звонка, и я знаю это, но я тоже увлекся или еще не осознал этого.

Я бы предпочел написать код для решения этой проблемы, и создавать этот объект было немного интересно.

Этот объект также является разновидностью подкласса в форме оболочки. Вы можете обернуть экземпляр объекта экземпляром объекта для достижения такого же эффекта, как если бы вы могли создать подкласс класса. Подклассификация не является жизнеспособным решением для такого API, как Tweepy, потому что слишком много движущихся частей, которые нужно покрыть. После обертывания экземпляра Tweepy API его можно рассматривать как оборачиваемый объект. Таким образом, это означает, что показанный выше объект является прокси для обернутого объекта, потому что он перехватывает запросы через __call__, которые он затем передает после выполнения некоторой работы по дому, чтобы гарантировать, что ограничения скорости не нарушаются.

Вы могли заметить, что я использую eval, и я знаю, что это может вызвать недоумение в определенных кругах, но мне нравится eval, потому что он предоставляет компилятор Python, и мне нравится использовать открытые вещи. Если вы читали другие мои статьи, возможно, вы знаете, что мне нравится решать проблемы с рабочим кодом независимо от того, как он был написан. Пока код работает, я умею им пользоваться. Если код не работает, я исправляю его, и меня не волнует, как это было исправлено, или методы, которые я использовал, чтобы исправить это.

Иногда я беру интервью у людей, которые, кажется, считают, что нужно знать весь фреймворк или библиотеку, чтобы быть эффективным, но это не так. Вы будете удивлены, узнав, как мало мне нужно знать о каком-либо предмете, чтобы получить хорошие воспроизводимые результаты от его использования.

В какой-то момент вы можете получить код здесь. Я часто совершаю коммиты, поэтому следите за обновлениями, если вы не смогли найти код из этой статьи.