Как реализовать механизм повтора при отсутствии данных в обработчике веб-перехватчика? [дубликат] ⇐ Python

[Anonymous]

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

Код: Выделить всё

def handle_rocket_reach_webhook(self, profile_data: any):
try:
incoming_webhook_lookup_id = profile_data.get("id", None)

provider_webhook: Optional[ProviderWebhook] = self._provider_webhook_repo.get_one_by_filter(
filters={'lookup_id': int(incoming_webhook_lookup_id)})

if not provider_webhook:
logger.warning(f"No matching document found for lookup ID {incoming_webhook_lookup_id}.")
return {"message": "Document not found"}

# Further processing logic...
except Exception as e:
logger.error(f"Error in RocketReach webhook call: {e}")
return {"message": "Internal server error"}

Мне нужно добавить механизм повтора для случая, когда значение «input_webhook» не соответствует true. В частности, я хочу повторить логику через 1 час (максимум 3 попытки).
Как лучше всего реализовать это на Python? В настоящее время я использую FastAPI и BackgroundTasks, но при необходимости открыт для других предложений, таких как asyncio или планировщики задач, такие как Celery или APScheduler.
Я был бы признателен за пример кода или подробное объяснение. о том, как это реализовать.

Подробнее здесь: https://stackoverflow.com/questions/793 ... ok-handler