Вызывать дорогостоящий вызов API только один раз каждые n секунд

Существует дорогостоящий вызов API, который занимает пару секунд и записывает много данных. Я стараюсь ограничить количество звонков до одного раза в interval секунд. Проблема в том, что ЕСЛИ вызывается API, он обычно вызывается несколько раз одновременно.

Итак, что я сделал:

class LimitAPICalls:
    def decide_for_poll(self, interval):
        self.time_now = datetime.datetime.now()
        try:
            time_since_last_run = (self.time_now - self.time_old).seconds
        except Exception as E:
            time_since_last_run = interval

        if time_since_last_run >= interval:
            self.time_old = datetime.datetime.now()
            return True
        return False

    def call_wrapper(self):
        if self.decide_for_poll(interval = 30) == True:
            print("actually calling")
            self.result = the_call_itself()
        try:
            print("returning values from memory")
            return self.result
        except AttributeError:
            print("calling because of AttributeError")
            self.time_old = datetime.datetime.now()
            return the_call_itself()

То, что я вижу в консоли, особенно при повторной инициализации, более или менее всегда представляет собой шаблон, похожий на:

actually calling 
returning values from memoryreturning values from memory
returning values from memorycalling because of AttributeError
calling because of AttributeError
calling because of AttributeError 
returning values from memory 
calling because of AttributeError    
returning values from memory returning values from memory
returning values from memory returning values from memoryreturning values from memory
returning values from memory

Но я хочу, чтобы первый вызов гарантировал, что на время interval будет сделан единственный вызов. Таким образом, вызов выполняется несколько раз, потому что он еще не завершен.

Звонки идут из браузера, все сразу - как сделать так, чтобы за время interval был сделан только один звонок?

Вам не хватает if-else... Вы делаете вызов, если интервал был достаточно длинным, затем вы делаете вызов снова.

Guimoute 10.04.2022 17:30

Не могли бы вы быть более точным (извините, возможно, вы правы на 100%, я просто смотрел на это некоторое время).

xtlc 10.04.2022 17:33

В call_wrapperthe_call_itself появляется три раза. Функция вызывается один раз, если в последний раз она вызывалась более <interval> назад, затем вы вызываете ее снова, независимо от условия в блоке try-except.

Guimoute 10.04.2022 17:34

простите. это была ошибка копирования и вставки на моей стороне. Ты был прав. Пожалуйста, посмотрите еще раз :) Проблема та же, что и раньше.

xtlc 10.04.2022 17:38

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

xtlc 10.04.2022 17:40
Анализ настроения постов в Twitter с помощью Python, Tweepy и Flair
Анализ настроения постов в Twitter с помощью Python, Tweepy и Flair
Анализ настроения текстовых сообщений может быть настолько сложным или простым, насколько вы его сделаете. Как и в любом ML-проекте, вы можете выбрать...
7 лайфхаков для начинающих Python-программистов
7 лайфхаков для начинающих Python-программистов
В этой статье мы расскажем о хитростях и советах по Python, которые должны быть известны разработчику Python.
Установка Apache Cassandra на Mac OS
Установка Apache Cassandra на Mac OS
Это краткое руководство по установке Apache Cassandra.
Сертификатная программа "Кванты Python": Бэктестер ансамблевых методов на основе ООП
Сертификатная программа "Кванты Python": Бэктестер ансамблевых методов на основе ООП
В одном из недавних постов я рассказал о том, как я использую навыки количественных исследований, которые я совершенствую в рамках программы TPQ...
Создание персонального файлового хранилища
Создание персонального файлового хранилища
Вы когда-нибудь хотели поделиться с кем-то файлом, но он содержал конфиденциальную информацию? Многие думают, что электронная почта безопасна, но это...
Создание приборной панели для анализа данных на GCP - часть I
Создание приборной панели для анализа данных на GCP - часть I
Недавно я столкнулся с интересной бизнес-задачей - визуализацией сбоев в цепочке поставок лекарств, которую могут просматривать врачи и...
0
5
32
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Можешь попробовать с этим? Я удалил все ваши блоки try-except, потому что они не нужны.


class LimitAPICalls:

    def __init__(self): 
        self.time_old = datetime.datetime(1, 1, 1) # A long time ago so that the first comparison is always true.
        self.result = None

    def poll(self):
        return NotImplemented # The costly API request goes here.

    def decide_for_poll(self, interval) -> bool:      
        time_now = datetime.datetime.now()
        time_since_last_run = (time_now - self.time_old).seconds
        if time_since_last_run >= interval:
            self.time_old = datetime.datetime.now()
            return True
        return False

    def call_wrapper(self):
        if self.decide_for_poll(interval=30) == True:
            print("actually calling")
            self.result = self.poll()
        else:
            print("returning values from memory")
        return self.result 

в переменной time_now есть небольшая ошибка, которая в любом случае устарела. Но это помогает, спасибо!

xtlc 10.04.2022 18:28

это иногда терпит неудачу между прочим... возвращается пустой результат, и вызов не выполняется.

xtlc 10.04.2022 18:36

Я думаю, никто не сможет вам помочь, не зная содержания вызова и того, как вы запускаете этот код. Это основано на requests?

Guimoute 10.04.2022 19:13

да, это так ... пока я просто решил это, проверив вывод self.result, и если тип None, я повторяю процесс.

xtlc 10.04.2022 20:54

Другие вопросы по теме