prefect work-pool update --base-job-template base-job-template.json my-work-pool

from typing import Any, Dict

from prefect import get_client
from prefect.client.schemas.actions import WorkPoolCreate, WorkPoolUpdate
from prefect.exceptions import ObjectAlreadyExists

async def create_update_work_pool(work_pool_name: str, job_template: Dict[str, Any]) -> None:
    """Create or update a Prefect Work Pool

    Args:
        work_pool_name (str): name of the work pool to create/update
        job_template (Dict[str, Any]: work pool job template to create/update
    """
    async with get_client() as client:  # type: ignore [union-attr]
        try:
            await client.create_work_pool(
                WorkPoolCreate(
                    name=work_pool_name,
                    type="kubernetes",
                    base_job_template=job_template,
                    description="",
                )
            )
        except ObjectAlreadyExists:
            await client.update_work_pool(
                work_pool_name=work_pool_name,
                work_pool=WorkPoolUpdate(
                    base_job_template=job_template,
                    is_paused=False,
                    description="",
                    concurrency_limit=None,
                ),
            )

import asyncio
import json

with open("base_job_template.json", "r") as f:
    job_template = json.load(f)
asyncio.run(create_update_work_pool("test_work_pool", job_template)

--overwrite