Code cleaning

Move trigger_ztp_generation and trigger_vm_creation from tasks.py to
utils.py.
The tasks.py module should only include functions used by the worker.

app/network/views.py

@@ -38,7 +38,7 @@ from .forms import (
 )
 from ..extensions import db
 from ..decorators import login_groups_accepted
-from .. import models, utils, tasks
+from .. import models, utils
 
 bp = Blueprint("network", __name__)
 
@@ -167,7 +167,7 @@ def view_host(name):
                 )
                 return redirect(url_for("network.view_host", name=name))
             else:
-                task = tasks.trigger_ztp_configuration(host)
+                task = utils.trigger_ztp_configuration(host)
                 db.session.commit()
                 current_app.logger.info(
                     f"Generation of {name} ZTP configuration requested: task {task.id}"
@@ -185,7 +185,7 @@ def view_host(name):
                 )
                 return redirect(url_for("network.view_host", name=name))
             else:
-                task = tasks.trigger_vm_creation(
+                task = utils.trigger_vm_creation(
                     host,
                     vm_disk_size=int(form.disk.data),
                     vm_cores=int(form.cores.data),

app/tasks.py

@@ -16,7 +16,6 @@ import requests
 import tower_cli
 from datetime import datetime
 from flask import current_app
-from flask_login import current_user
 from rq import Worker, get_current_job
 from openpyxl import Workbook
 from .extensions import db
@@ -83,87 +82,6 @@ class TaskWorker(Worker):
         super().prepare_job_execution(job)
 
 
-def trigger_vm_creation(
-    host, vm_disk_size, vm_cores, vm_memory, vm_osversion, skip_post_install_job
-):
-    """Trigger a job to create a virtual machine or virtual IOC"""
-    domain = str(host.main_interface.network.domain)
-    extra_vars = [
-        f"vmname={host.name}",
-        f"domain={domain}",
-        f"vm_disk_size={vm_disk_size}",
-        f"vm_cores={vm_cores}",
-        f"vm_memory={vm_memory}",
-        f"vm_osversion={vm_osversion}",
-    ]
-    if host.is_ioc:
-        task_name = "trigger_vioc_creation"
-        job_template = current_app.config["AWX_CREATE_VIOC"]
-        post_job_template = current_app.config["AWX_POST_INSTALL"]["VIOC"].get(domain)
-    else:
-        task_name = "trigger_vm_creation"
-        job_template = current_app.config["AWX_CREATE_VM"]
-        post_job_template = current_app.config["AWX_POST_INSTALL"]["VM"].get(domain)
-    current_app.logger.info(
-        f"Launch new job to create the {host} VM: {job_template} with {extra_vars}"
-    )
-    task = current_user.launch_task(
-        task_name,
-        resource="job",
-        queue_name="low",
-        func="launch_awx_job",
-        job_template=job_template,
-        extra_vars=extra_vars,
-        timeout=500,
-    )
-    if (
-        post_job_template
-        and (not skip_post_install_job)
-        and (not vm_osversion.startswith("windows"))
-    ):
-        current_user.launch_task(
-            "trigger_post_install_job",
-            resource="job",
-            queue_name="low",
-            func="launch_awx_job",
-            job_template=post_job_template,
-            limit=f"{host.fqdn}",
-            depends_on=task.id,
-            timeout=500,
-        )
-        current_app.logger.info(
-            f"Trigger post install job: run {post_job_template} on {host.fqdn}"
-        )
-    return task
-
-
-def trigger_ztp_configuration(host):
-    """Trigger a job to generate a switch ZTP configuration"""
-    extra_vars = [
-        f"ztp_host={host.name}",
-        f"ztp_ip={host.main_interface.ip}",
-        f"ztp_mac={host.main_interface.mac}",
-        f"ztp_model={host.model}",
-    ]
-    ztp_stack = ",".join([item.serial_number for item in host.stack_members()])
-    if ztp_stack:
-        extra_vars.append(f"ztp_stack={ztp_stack}")
-    job_template = current_app.config["AWX_ZTP_CONFIGURATION"]
-    current_app.logger.info(
-        f"Launch new job to generate ZTP configuration for {host.name} device: {job_template} with {extra_vars}"
-    )
-    task = current_user.launch_task(
-        "trigger_ztp_configuration",
-        resource="job",
-        queue_name="low",
-        func="launch_awx_job",
-        job_template=job_template,
-        extra_vars=extra_vars,
-        timeout=500,
-    )
-    return task
-
-
 def launch_awx_job(resource="job", **kwargs):
     """Launch an AWX job
 

app/utils.py

@@ -319,6 +319,87 @@ def trigger_ioc_repository_creation(host):
         return task
 
 
+def trigger_vm_creation(
+    host, vm_disk_size, vm_cores, vm_memory, vm_osversion, skip_post_install_job
+):
+    """Trigger a job to create a virtual machine or virtual IOC"""
+    domain = str(host.main_interface.network.domain)
+    extra_vars = [
+        f"vmname={host.name}",
+        f"domain={domain}",
+        f"vm_disk_size={vm_disk_size}",
+        f"vm_cores={vm_cores}",
+        f"vm_memory={vm_memory}",
+        f"vm_osversion={vm_osversion}",
+    ]
+    if host.is_ioc:
+        task_name = "trigger_vioc_creation"
+        job_template = current_app.config["AWX_CREATE_VIOC"]
+        post_job_template = current_app.config["AWX_POST_INSTALL"]["VIOC"].get(domain)
+    else:
+        task_name = "trigger_vm_creation"
+        job_template = current_app.config["AWX_CREATE_VM"]
+        post_job_template = current_app.config["AWX_POST_INSTALL"]["VM"].get(domain)
+    current_app.logger.info(
+        f"Launch new job to create the {host} VM: {job_template} with {extra_vars}"
+    )
+    task = current_user.launch_task(
+        task_name,
+        resource="job",
+        queue_name="low",
+        func="launch_awx_job",
+        job_template=job_template,
+        extra_vars=extra_vars,
+        timeout=500,
+    )
+    if (
+        post_job_template
+        and (not skip_post_install_job)
+        and (not vm_osversion.startswith("windows"))
+    ):
+        current_user.launch_task(
+            "trigger_post_install_job",
+            resource="job",
+            queue_name="low",
+            func="launch_awx_job",
+            job_template=post_job_template,
+            limit=f"{host.fqdn}",
+            depends_on=task.id,
+            timeout=500,
+        )
+        current_app.logger.info(
+            f"Trigger post install job: run {post_job_template} on {host.fqdn}"
+        )
+    return task
+
+
+def trigger_ztp_configuration(host):
+    """Trigger a job to generate a switch ZTP configuration"""
+    extra_vars = [
+        f"ztp_host={host.name}",
+        f"ztp_ip={host.main_interface.ip}",
+        f"ztp_mac={host.main_interface.mac}",
+        f"ztp_model={host.model}",
+    ]
+    ztp_stack = ",".join([item.serial_number for item in host.stack_members()])
+    if ztp_stack:
+        extra_vars.append(f"ztp_stack={ztp_stack}")
+    job_template = current_app.config["AWX_ZTP_CONFIGURATION"]
+    current_app.logger.info(
+        f"Launch new job to generate ZTP configuration for {host.name} device: {job_template} with {extra_vars}"
+    )
+    task = current_user.launch_task(
+        "trigger_ztp_configuration",
+        resource="job",
+        queue_name="low",
+        func="launch_awx_job",
+        job_template=job_template,
+        extra_vars=extra_vars,
+        timeout=500,
+    )
+    return task
+
+
 def redirect_to_job_status(job_id):
     """
     The answer to a client request, leading it to regularly poll a job status.