prefect_utils.py

import subprocess
import threading

def start_prefect_worker(work_pool_name: str = "ot2-pool"):
    def worker_runner():
        print(f"Starting Prefect Worker for work pool: {work_pool_name}")
        subprocess.run([
            "prefect", "worker", "start",
            "--pool", work_pool_name
        ])
    worker_thread = threading.Thread(target=worker_runner, daemon=True)
    worker_thread.start()
    print("Prefect Worker started in background thread.")


from prefect.client import get_client

async def trigger_maintenance_request(maintenance_type: str):
    async with get_client() as client:
        # find deployment id
        deployment = await client.read_deployment_by_name(
            name="request-wells-maintenance/wells-maintenance"
        )

        await client.create_flow_run_from_deployment(
            deployment_id=deployment.id,
            parameters={"maintenance_type": maintenance_type}
        )