NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-27 10:58:37 -06:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Enqueue a SyncDataSourceJob when needed after saving a DataSource

Browse Source
This commit is contained in:
Jeremy Stretch 2025-02-26 15:55:02 -05:00
parent 5e9172ce97
commit 91d8cb596e
1 changed files with 15 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
netbox/core/signals.py
View File

@ -10,14 +10,13 @@ from django_prometheus.models import model_deletes, model_inserts, model_updates
from core.choices import ObjectChangeActionChoices from core.choices import ObjectChangeActionChoices
from core.events import * from core.events import *
from core.models import ObjectChange
from extras.events import enqueue_event from extras.events import enqueue_event
from extras.utils import run_validators from extras.utils import run_validators
from netbox.config import get_config from netbox.config import get_config
from netbox.context import current_request, events_queue from netbox.context import current_request, events_queue
from netbox.models.features import ChangeLoggingMixin from netbox.models.features import ChangeLoggingMixin
from utilities.exceptions import AbortRequest from utilities.exceptions import AbortRequest
from .models import ConfigRevision from .models import ConfigRevision, DataSource, ObjectChange
__all__ = ( __all__ = (
'clear_events', 'clear_events',
@ -182,6 +181,20 @@ def clear_events_queue(sender, **kwargs):
# DataSource handlers # DataSource handlers
# #
@receiver(post_save, sender=DataSource)
def enqueue_sync_job(instance, **kwargs):
"""
When a DataSource is saved, check its sync_interval and enqueue a sync job if appropriate.
"""
from .jobs import SyncDataSourceJob
if instance.sync_interval:
SyncDataSourceJob.enqueue_once(instance, interval=instance.sync_interval)
else:
# Delete any previously scheduled recurring jobs for this DataSource
SyncDataSourceJob.get_jobs(instance).filter(sync_interval__isnull=False).delete()
@receiver(post_sync) @receiver(post_sync)
def auto_sync(instance, **kwargs): def auto_sync(instance, **kwargs):
""" """