37 lines
1.0 KiB
Python
37 lines
1.0 KiB
Python
from __future__ import annotations
|
|
|
|
from core.exceptions import JobFailed
|
|
from netbox.jobs import JobRunner
|
|
|
|
from .lifecycle import LifecycleRequest, build_service
|
|
|
|
|
|
class PluginLifecycleJob(JobRunner):
|
|
"""Background execution is intentionally restricted to non-mutating plans."""
|
|
|
|
class Meta:
|
|
name = "Plugin Store dry-run"
|
|
|
|
def run(
|
|
self,
|
|
*,
|
|
audit_id: int,
|
|
slug: str,
|
|
action: str,
|
|
version: str = "",
|
|
dry_run: bool = True,
|
|
actor_id: int | None = None,
|
|
**kwargs,
|
|
):
|
|
if not dry_run:
|
|
raise JobFailed("Mutating lifecycle actions may not run inside netbox-rq.")
|
|
try:
|
|
build_service().execute(
|
|
LifecycleRequest(slug=slug, action=action, version=version, dry_run=True),
|
|
actor_id=actor_id,
|
|
audit_id=audit_id,
|
|
)
|
|
except Exception as exc:
|
|
self.logger.error("Plugin Store dry-run failed: %s", exc)
|
|
raise JobFailed(str(exc)) from exc
|