Code: Select all
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"}
Code: Select all
scheduler = BackgroundScheduler()
scheduler.add_job(
method_name,
trigger=IntervalTrigger(seconds=10),
args=[profile_data, scheduler],
id="limited_job",
replace_existing=True,
)
Wo werden die Verzögerungsinformationen gespeichert? im RAM (da es sich im Arbeitsspeicher befindet)
oder woanders
Wenn ja, benötige ich Dokumentation.