Use a decorator to register system jobs

This commit is contained in:
Jeremy Stretch 2024-10-30 13:40:30 -04:00
parent 2bd45d3343
commit 566201507d
5 changed files with 21 additions and 21 deletions

View File

@ -9,12 +9,25 @@ from rq.timeouts import JobTimeoutException
from core.choices import JobStatusChoices
from core.models import Job, ObjectType
from netbox.constants import ADVISORY_LOCK_KEYS
from netbox.registry import registry
__all__ = (
'JobRunner',
'system_job',
)
def system_job():
"""
Decorator for registering a `JobRunner` class as system background job.
"""
def _wrapper(cls):
registry['system_jobs'][cls.name] = cls
return cls
return _wrapper
class JobRunner(ABC):
"""
Background Job helper class.

View File

@ -8,7 +8,7 @@ from packaging import version
from netbox.registry import registry
from netbox.search import register_search
from netbox.utils import register_data_backend, register_system_job
from netbox.utils import register_data_backend
from .navigation import *
from .registration import *
from .templates import *
@ -26,7 +26,6 @@ registry['plugins'].update({
DEFAULT_RESOURCE_PATHS = {
'search_indexes': 'search.indexes',
'data_backends': 'data_backends.backends',
'system_jobs': 'jobs.system_jobs',
'graphql_schema': 'graphql.schema',
'menu': 'navigation.menu',
'menu_items': 'navigation.menu_items',
@ -74,7 +73,6 @@ class PluginConfig(AppConfig):
# Optional plugin resources
search_indexes = None
data_backends = None
system_jobs = None
graphql_schema = None
menu = None
menu_items = None
@ -114,11 +112,6 @@ class PluginConfig(AppConfig):
for backend in data_backends:
register_data_backend()(backend)
# Register system jobs (if defined)
system_jobs = self._load_resource('system_jobs') or []
for job in system_jobs:
register_system_job()(job)
# Register template content (if defined)
if template_extensions := self._load_resource('template_extensions'):
register_template_extensions(template_extensions)

View File

@ -21,5 +21,10 @@ class DummyPluginConfig(PluginConfig):
'netbox.tests.dummy_plugin.events.process_events_queue'
]
def ready(self):
super().ready()
from . import jobs # noqa: F401
config = DummyPluginConfig

View File

@ -1,7 +1,8 @@
from core.choices import JobIntervalChoices
from netbox.jobs import JobRunner
from netbox.jobs import JobRunner, system_job
@system_job()
class DummySystemJob(JobRunner):
class Meta:
system_interval = JobIntervalChoices.INTERVAL_HOURLY

View File

@ -3,7 +3,6 @@ from netbox.registry import registry
__all__ = (
'get_data_backend_choices',
'register_data_backend',
'register_system_job',
)
@ -25,14 +24,3 @@ def register_data_backend():
return cls
return _wrapper
def register_system_job():
"""
Decorator for registering a `JobRunner` class as system background job.
"""
def _wrapper(cls):
registry['system_jobs'][cls.name] = cls
return cls
return _wrapper